max / makenotwork

18.0 KB · 569 lines History Blame Raw

1	//! HTTP API for programmatic ticket management and peer sync.
2	//!
3	//! The API runs on the tailnet, but tailnet membership is not sufficient on its
4	//! own: anything that can reach the port could otherwise forge tickets. So the
5	//! server fails closed -- it never serves without a shared bearer token. The
6	//! token is resolved in order from `--token`, the `WAM_TOKEN` env var, a token
7	//! persisted beside the database, or, failing all of those, a freshly generated
8	//! one that is persisted and printed once at startup. Every request (including
9	//! peer sync) must carry `Authorization: Bearer <token>`.
10
11	use std::sync::Arc;
12
13	use color_eyre::eyre::WrapErr;
14	use tokio::sync::Mutex;
15
16	use axum::{
17	Json, Router,
18	extract::{Path, Query, Request, State},
19	http::StatusCode,
20	middleware::{self, Next},
21	response::{IntoResponse, Response},
22	routing::{get, patch, post},
23	};
24	use rusqlite::Connection;
25	use serde::Deserialize;
26
27	use crate::db::{self, ListFilter};
28	use crate::types::{Channel, NewTicket, Priority, Status, Ticket};
29
30	/// Shared state: SQLite connection + node identity + shared bearer token.
31	#[derive(Clone)]
32	pub(crate) struct AppState {
33	pub db: Arc<Mutex<Connection>>,
34	pub node_id: String,
35	/// Shared bearer token required on every request and presented to peers.
36	/// Always present -- the server refuses to start without one.
37	pub token: Arc<str>,
38	}
39
40	/// Start the HTTP server, optionally syncing with peers.
41	///
42	/// `token` is the CLI-supplied shared secret. Token resolution fails closed:
43	/// see [`resolve_or_create_token`]. Every request (including peer sync) must
44	/// present the resolved token as a bearer token.
45	pub(crate) async fn serve(
46	conn: Connection,
47	port: u16,
48	peers: Vec<String>,
49	token: Option<String>,
50	) -> color_eyre::eyre::Result<()> {
51	let node_id = db::get_or_create_node_id(&conn)?;
52	eprintln!("node: {}", &node_id[..8]);
53
54	let token = resolve_or_create_token(token)?;
55
56	let app_state = AppState {
57	db: Arc::new(Mutex::new(conn)),
58	node_id,
59	token: Arc::clone(&token),
60	};
61
62	// Spawn sync loop if peers are configured
63	if !peers.is_empty() {
64	let sync_state = app_state.clone();
65	let sync_peers = peers.clone();
66	tokio::spawn(async move {
67	sync_loop(sync_state, sync_peers).await;
68	});
69	}
70
71	let app = Router::new()
72	.route("/tickets", post(create_ticket))
73	.route("/tickets", get(list_tickets))
74	.route("/tickets/{id}", get(get_ticket))
75	.route("/tickets/{id}", patch(update_ticket))
76	.route("/sync/pull", get(sync_pull))
77	.route("/sync/push", post(sync_push))
78	.route("/sync/node", get(sync_node_info))
79	.route_layer(middleware::from_fn_with_state(
80	app_state.clone(),
81	require_auth,
82	))
83	.with_state(app_state);
84
85	let addr = format!("0.0.0.0:{port}");
86	let listener = tokio::net::TcpListener::bind(&addr).await?;
87	eprintln!("wam serving on {addr}");
88	eprintln!("auth: bearer token required on every request");
89	if !peers.is_empty() {
90	eprintln!("syncing with {} peer(s)", peers.len());
91	}
92	axum::serve(listener, app).await?;
93	Ok(())
94	}
95
96	/// Resolve the shared bearer token, failing closed so the API is never open.
97	///
98	/// Precedence: the explicit `--token` flag, then the `WAM_TOKEN` env var, then a
99	/// token persisted beside the database from a previous run, then a freshly
100	/// generated one (persisted with `0600` perms and printed once). The result is
101	/// always a usable token; there is no unauthenticated path.
102	fn resolve_or_create_token(cli_token: Option<String>) -> color_eyre::eyre::Result<Arc<str>> {
103	// An explicit flag or environment variable wins, in that order.
104	let supplied = cli_token
105	.or_else(\|\| std::env::var("WAM_TOKEN").ok())
106	.map(\|t\| t.trim().to_owned())
107	.filter(\|t\| !t.is_empty());
108	if let Some(tok) = supplied {
109	return Ok(Arc::from(tok));
110	}
111
112	token_from_dir(&db::data_dir()?)
113	}
114
115	/// Reuse a token persisted in `dir/token`, or mint, persist, and print a new
116	/// one. Split out from [`resolve_or_create_token`] so the disk paths are
117	/// testable against a scratch directory.
118	fn token_from_dir(dir: &std::path::Path) -> color_eyre::eyre::Result<Arc<str>> {
119	let path = dir.join("token");
120
121	// Reuse a token persisted by an earlier run.
122	if let Ok(saved) = std::fs::read_to_string(&path) {
123	let saved = saved.trim().to_owned();
124	if !saved.is_empty() {
125	eprintln!("auth: using persisted token at {}", path.display());
126	return Ok(Arc::from(saved));
127	}
128	}
129
130	// Nothing configured: mint one, persist it, and print it once so the
131	// operator can propagate it to producers and peers.
132	let token = format!(
133	"{}{}",
134	uuid::Uuid::new_v4().simple(),
135	uuid::Uuid::new_v4().simple()
136	);
137	write_token_file(&path, &token)?;
138	eprintln!(
139	"auth: no token supplied -- generated one and saved it to {}",
140	path.display()
141	);
142	eprintln!(" token: {token}");
143	eprintln!(" set WAM_TOKEN to this value on every producer and peer.");
144	Ok(Arc::from(token))
145	}
146
147	/// Write the token to `path`, restricting it to owner-only (`0600`) on unix.
148	fn write_token_file(path: &std::path::Path, token: &str) -> color_eyre::eyre::Result<()> {
149	std::fs::write(path, format!("{token}\n"))
150	.wrap_err_with(\|\| format!("write token file: {}", path.display()))?;
151	#[cfg(unix)]
152	{
153	use std::os::unix::fs::PermissionsExt;
154	std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
155	.wrap_err_with(\|\| format!("restrict token file perms: {}", path.display()))?;
156	}
157	Ok(())
158	}
159
160	/// Reject any request lacking a valid `Authorization: Bearer <token>` header.
161	/// The token is always set, so every route behind this layer is authenticated.
162	async fn require_auth(State(state): State<AppState>, req: Request, next: Next) -> Response {
163	let expected = state.token.as_ref();
164	let presented = req
165	.headers()
166	.get(axum::http::header::AUTHORIZATION)
167	.and_then(\|v\| v.to_str().ok())
168	.and_then(\|v\| v.strip_prefix("Bearer "));
169	match presented {
170	Some(tok) if ct_eq(tok.as_bytes(), expected.as_bytes()) => next.run(req).await,
171	_ => (
172	StatusCode::UNAUTHORIZED,
173	Json(serde_json::json!({"error": "unauthorized"})),
174	)
175	.into_response(),
176	}
177	}
178
179	/// Constant-time byte comparison, so a wrong token can't be recovered by timing
180	/// how far it matched.
181	fn ct_eq(a: &[u8], b: &[u8]) -> bool {
182	if a.len() != b.len() {
183	return false;
184	}
185	let mut diff = 0u8;
186	for (x, y) in a.iter().zip(b) {
187	diff \|= x ^ y;
188	}
189	diff == 0
190	}
191
192	#[cfg(test)]
193	mod tests {
194	use super::*;
195
196	#[test]
197	fn ct_eq_matches_identical() {
198	assert!(ct_eq(b"s3cr3t-token", b"s3cr3t-token"));
199	assert!(ct_eq(b"", b""));
200	}
201
202	#[test]
203	fn ct_eq_rejects_different_content() {
204	assert!(!ct_eq(b"s3cr3t-token", b"s3cr3t-toker"));
205	assert!(!ct_eq(b"abc", b"xyz"));
206	}
207
208	#[test]
209	fn ct_eq_rejects_length_mismatch() {
210	// A correct prefix must not pass — the guard is exact-match only.
211	assert!(!ct_eq(b"s3cr3t", b"s3cr3t-token"));
212	assert!(!ct_eq(b"s3cr3t-token", b"s3cr3t"));
213	}
214
215	#[test]
216	fn with_token_attaches_only_when_set() {
217	let client = reqwest::Client::new();
218	// Smoke: both branches build a valid request without panicking.
219	let _ = with_token(client.get("http://127.0.0.1/x"), Some("tok"));
220	let _ = with_token(client.get("http://127.0.0.1/x"), None);
221	}
222
223	/// A unique scratch directory under the system temp dir, removed on drop.
224	struct ScratchDir(std::path::PathBuf);
225
226	impl ScratchDir {
227	fn new() -> Self {
228	let dir = std::env::temp_dir().join(format!("wam-test-{}", uuid::Uuid::new_v4()));
229	std::fs::create_dir_all(&dir).unwrap();
230	Self(dir)
231	}
232	}
233
234	impl Drop for ScratchDir {
235	fn drop(&mut self) {
236	let _ = std::fs::remove_dir_all(&self.0);
237	}
238	}
239
240	#[test]
241	fn explicit_token_wins_and_is_trimmed() {
242	// A supplied token short-circuits before any env or disk lookup.
243	let tok = resolve_or_create_token(Some(" s3cr3t ".to_string())).unwrap();
244	assert_eq!(&*tok, "s3cr3t");
245	}
246
247	#[test]
248	fn token_from_dir_generates_persists_and_reuses() {
249	let scratch = ScratchDir::new();
250
251	// First call mints a token and writes it to dir/token.
252	let first = token_from_dir(&scratch.0).unwrap();
253	assert!(!first.is_empty());
254	let path = scratch.0.join("token");
255	assert!(path.exists());
256	assert_eq!(std::fs::read_to_string(&path).unwrap().trim(), &*first);
257
258	// On unix the file is owner-only.
259	#[cfg(unix)]
260	{
261	use std::os::unix::fs::PermissionsExt;
262	let mode = std::fs::metadata(&path).unwrap().permissions().mode();
263	assert_eq!(mode & 0o777, 0o600);
264	}
265
266	// A second call reuses the persisted token rather than minting anew.
267	let second = token_from_dir(&scratch.0).unwrap();
268	assert_eq!(&first, &second);
269	}
270
271	#[test]
272	fn token_from_dir_ignores_blank_file() {
273	let scratch = ScratchDir::new();
274	std::fs::write(scratch.0.join("token"), " \n").unwrap();
275	// A whitespace-only file is treated as absent: a real token is minted.
276	let tok = token_from_dir(&scratch.0).unwrap();
277	assert!(!tok.trim().is_empty());
278	assert_eq!(tok.len(), 64); // two hyphen-free v4 UUIDs
279	}
280	}
281
282	// -- Ticket handlers ----------------------------------------------------------
283
284	/// POST /tickets
285	async fn create_ticket(
286	State(state): State<AppState>,
287	Json(new): Json<NewTicket>,
288	) -> impl IntoResponse {
289	let conn = state.db.lock().await;
290	match db::create_ticket(&conn, &new, &state.node_id) {
291	Ok(ticket) => (StatusCode::CREATED, Json(serde_json::json!(ticket))).into_response(),
292	Err(e) => (
293	StatusCode::INTERNAL_SERVER_ERROR,
294	Json(serde_json::json!({"error": e.to_string()})),
295	)
296	.into_response(),
297	}
298	}
299
300	#[derive(Debug, Deserialize, Default)]
301	pub(crate) struct ListQuery {
302	pub status: Option<String>,
303	pub priority: Option<String>,
304	pub channel: Option<String>,
305	pub source: Option<String>,
306	pub search: Option<String>,
307	}
308
309	/// GET /tickets
310	async fn list_tickets(
311	State(state): State<AppState>,
312	Query(q): Query<ListQuery>,
313	) -> impl IntoResponse {
314	let status = q.status.as_deref().and_then(\|s\| s.parse::<Status>().ok());
315	let priority = q
316	.priority
317	.as_deref()
318	.and_then(\|s\| s.parse::<Priority>().ok());
319	let channel = q.channel.as_deref().and_then(\|s\| s.parse::<Channel>().ok());
320
321	let conn = state.db.lock().await;
322	let filter = ListFilter {
323	status,
324	priority,
325	channel,
326	source: q.source.as_deref(),
327	search: q.search.as_deref(),
328	};
329	match db::list_tickets(&conn, &filter) {
330	Ok(tickets) => {
331	Json(serde_json::json!({"data": tickets, "count": tickets.len()})).into_response()
332	}
333	Err(e) => (
334	StatusCode::INTERNAL_SERVER_ERROR,
335	Json(serde_json::json!({"error": e.to_string()})),
336	)
337	.into_response(),
338	}
339	}
340
341	/// GET /tickets/:id
342	async fn get_ticket(State(state): State<AppState>, Path(id): Path<String>) -> impl IntoResponse {
343	let conn = state.db.lock().await;
344	match db::get_ticket(&conn, &id) {
345	Ok(ticket) => Json(serde_json::json!(ticket)).into_response(),
346	Err(_) => (
347	StatusCode::NOT_FOUND,
348	Json(serde_json::json!({"error": "ticket not found"})),
349	)
350	.into_response(),
351	}
352	}
353
354	#[derive(Debug, Deserialize)]
355	pub(crate) struct UpdateBody {
356	pub status: Option<String>,
357	}
358
359	/// PATCH /tickets/:id
360	async fn update_ticket(
361	State(state): State<AppState>,
362	Path(id): Path<String>,
363	Json(body): Json<UpdateBody>,
364	) -> impl IntoResponse {
365	let conn = state.db.lock().await;
366
367	let Ok(ticket) = db::get_ticket(&conn, &id) else {
368	return (
369	StatusCode::NOT_FOUND,
370	Json(serde_json::json!({"error": "ticket not found"})),
371	)
372	.into_response();
373	};
374
375	if let Some(status_str) = body.status {
376	let Ok(status) = status_str.parse::<Status>() else {
377	return (
378	StatusCode::BAD_REQUEST,
379	Json(serde_json::json!({"error": format!("invalid status: {status_str}")})),
380	)
381	.into_response();
382	};
383	if let Err(e) = db::update_status(&conn, &ticket.id, status) {
384	return (
385	StatusCode::INTERNAL_SERVER_ERROR,
386	Json(serde_json::json!({"error": e.to_string()})),
387	)
388	.into_response();
389	}
390	}
391
392	match db::get_ticket(&conn, &ticket.id) {
393	Ok(t) => Json(serde_json::json!(t)).into_response(),
394	Err(e) => (
395	StatusCode::INTERNAL_SERVER_ERROR,
396	Json(serde_json::json!({"error": e.to_string()})),
397	)
398	.into_response(),
399	}
400	}
401
402	// -- Sync endpoints -----------------------------------------------------------
403
404	#[derive(Debug, Deserialize)]
405	pub(crate) struct SyncPullQuery {
406	/// RFC3339 timestamp. Returns tickets updated after this time.
407	pub since: String,
408	}
409
410	/// GET /sync/pull?since=<rfc3339> — peer pulls tickets updated after timestamp
411	async fn sync_pull(
412	State(state): State<AppState>,
413	Query(q): Query<SyncPullQuery>,
414	) -> impl IntoResponse {
415	let conn = state.db.lock().await;
416	match db::tickets_since(&conn, &q.since) {
417	Ok(tickets) => Json(serde_json::json!({
418	"tickets": tickets,
419	"count": tickets.len(),
420	"node_id": state.node_id,
421	}))
422	.into_response(),
423	Err(e) => (
424	StatusCode::INTERNAL_SERVER_ERROR,
425	Json(serde_json::json!({"error": e.to_string()})),
426	)
427	.into_response(),
428	}
429	}
430
431	/// POST /sync/push — peer pushes tickets to us
432	async fn sync_push(
433	State(state): State<AppState>,
434	Json(tickets): Json<Vec<Ticket>>,
435	) -> impl IntoResponse {
436	let conn = state.db.lock().await;
437	let mut accepted = 0u32;
438	let mut rejected = 0u32;
439
440	for ticket in &tickets {
441	match db::upsert_synced_ticket(&conn, ticket) {
442	Ok(true) => accepted += 1,
443	Ok(false) => rejected += 1,
444	Err(e) => {
445	eprintln!("sync upsert error for {}: {e}", ticket.short_id());
446	rejected += 1;
447	}
448	}
449	}
450
451	Json(serde_json::json!({
452	"accepted": accepted,
453	"rejected": rejected,
454	}))
455	}
456
457	/// GET /sync/node — returns this node's identity
458	async fn sync_node_info(State(state): State<AppState>) -> impl IntoResponse {
459	Json(serde_json::json!({
460	"node_id": state.node_id,
461	}))
462	}
463
464	/// Attach the shared bearer token to an outbound peer request when one is set.
465	fn with_token(req: reqwest::RequestBuilder, token: Option<&str>) -> reqwest::RequestBuilder {
466	match token {
467	Some(t) => req.bearer_auth(t),
468	None => req,
469	}
470	}
471
472	// -- Background sync loop -----------------------------------------------------
473
474	/// Periodically pull from all peers and push local changes.
475	async fn sync_loop(state: AppState, peers: Vec<String>) {
476	let client = reqwest::Client::builder()
477	.timeout(std::time::Duration::from_secs(10))
478	.connect_timeout(std::time::Duration::from_secs(5))
479	.build()
480	.unwrap();
481
482	loop {
483	tokio::time::sleep(std::time::Duration::from_secs(30)).await;
484
485	for peer in &peers {
486	if let Err(e) = sync_with_peer(&state, &client, peer).await {
487	eprintln!("sync with {peer}: {e}");
488	}
489	}
490	}
491	}
492
493	/// Pull new tickets from a peer, then push our new tickets to them.
494	async fn sync_with_peer(
495	state: &AppState,
496	client: &reqwest::Client,
497	peer_url: &str,
498	) -> Result<(), Box<dyn std::error::Error>> {
499	let conn = state.db.lock().await;
500
501	// Get our cursor for this peer (default to epoch)
502	let cursor =
503	db::get_sync_cursor(&conn, peer_url)?.unwrap_or_else(\|\| "1970-01-01T00:00:00Z".to_string());
504
505	drop(conn); // Release lock before HTTP
506
507	// Pull from peer
508	let pull_url = format!(
509	"{peer_url}/sync/pull?since={}",
510	urlencoding::encode(&cursor)
511	);
512	let resp: serde_json::Value = with_token(client.get(&pull_url), Some(state.token.as_ref()))
513	.send()
514	.await?
515	.json()
516	.await?;
517
518	let tickets: Vec<Ticket> = serde_json::from_value(
519	resp.get("tickets")
520	.cloned()
521	.unwrap_or(serde_json::json!([])),
522	)?;
523
524	if !tickets.is_empty() {
525	let conn = state.db.lock().await;
526	let mut latest_updated = cursor.clone();
527
528	for ticket in &tickets {
529	db::upsert_synced_ticket(&conn, ticket)?;
530	let ts = ticket.updated_at.to_rfc3339();
531	if ts > latest_updated {
532	latest_updated = ts;
533	}
534	}
535
536	db::set_sync_cursor(&conn, peer_url, &latest_updated)?;
537	drop(conn);
538
539	eprintln!("sync: pulled {} ticket(s) from {peer_url}", tickets.len());
540	}
541
542	// Push our changes to peer (tickets updated since their last pull from us)
543	// We use the same cursor — they'll filter by last-writer-wins
544	let conn = state.db.lock().await;
545	let our_tickets = db::tickets_since(&conn, &cursor)?;
546	drop(conn);
547
548	if !our_tickets.is_empty() {
549	let push_url = format!("{peer_url}/sync/push");
550	let resp: serde_json::Value = with_token(
551	client.post(&push_url).json(&our_tickets),
552	Some(state.token.as_ref()),
553	)
554	.send()
555	.await?
556	.json()
557	.await?;
558	let accepted = resp
559	.get("accepted")
560	.and_then(serde_json::Value::as_u64)
561	.unwrap_or(0);
562	if accepted > 0 {
563	eprintln!("sync: pushed {accepted} ticket(s) to {peer_url}");
564	}
565	}
566
567	Ok(())
568	}
569