HTTP transport: CBOR wire messages, HttpRemote client, oxy-server crate with bearer-token pushes, oxy serve 750251b21a96

Author: Andy McCall @andymccall <[email protected]>
When: 1 month ago · 2026-09-26 18:47 UTC
Signature: ✔ signed by Andy McCall
Hash: 750251b21a96e27ed790cb825a7cc67a5967c07a749eae6f6a3f8a3361045beb
Depends on: 1983686fb7a7 20247a03a596 5b9c6ad45f6e e993a086c13f ff0af1ee299d
Depended on by: 835ceb9b3ea7 135bdcadec4a ed931d1d1102 0588de69a84d 569802ebae0e 4275ecb110f5 642fa213de80 80b069facc69 e02edccce98c e30c081a5064 b260109d27c5 f3811eb20bea d3b270909c78 b63fc7a77fc6 82202a4d0025 ee130dd9eef2 1ec6fdbc6bcb 6f06d9f912b4 67a4fe1aa2d5 028545dfa7a4 51d4862a16d8 4b9d06778581 68dcf21fe055 f05b84d1ad2d f53cbe7a204e f0a517d28237 efad40f9ac8e 68a924846379 e2a5edbf1395 cb8b8082fa73 d0bf2da15094 579afd107b9b c71a5777509a 2cd40ea1c04c f874dc8659c2 8d80356371c5 9f9e72b93e4e 6a0008781d5e bea00d4177c6 827281eed2ed b0461604ed98 14edbaaf5291 ff1f62fee654 2c75672fb2fe 598009a75f6c 6f8aeb996558 be451356e3ef 8c30d814f2cc f8c695764c95 11c958021e1a d42369abf180 3e5772ccf9da 37c299cec358 d82e44097323 be0161a88dd7 76f752503e7e 119bbf50cb86 7aa2dc9369e3 b0e64e81e4fb 05e71ed6cdf1 75167672ebcd e02bf0cd0533 c6aa5bd9d1c8 b3c5a59a875a 2b872d074ddb 28fc23d7706a 132b884b9a8d 2863456a0f6d 7421309f5372 135212de5370 d02e545dc265 83e0677b07fe 2b5ee371060e 982864068a59 fce44ae9ad29 02f1b14af1e2
In bookmarks: agents-in-readme bare-hashes carried-not-removed carry-identical doctor-trust identical-sides keyless-tags main merge-remote-bookmark only-help-103 paths-for-resolutions-and-amend push-notes show-stat subrepo-pin-state switch-says
In tags: v0.1.0 v0.10.0 v0.11.0 v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.4.1 v0.4.2 v0.4.3 v0.4.4 v0.4.5 v0.4.6 v0.5.0 v0.5.1 v0.6.0 v0.7.0 v0.8.0 v0.9.0
Edits: 71 atoms, 6 tree operations, 9 files

Browse the tree at this change

hide whitespace changes

added crates/oxy-server/Cargo.toml +14 −0

added crates/oxy-server/src/lib.rs +166 −0

added crates/oxy-server/tests/http.rs +89 −0

modified Cargo.lock +877 −6

modified Cargo.toml +2 −1

modified crates/oxy-core/Cargo.toml +1 −0

modified crates/oxy-core/src/protocol.rs +360 −6

modified crates/oxy-core/src/repo.rs +29 −7

modified src/main.rs +19 −1

1	+	[package]
2	+	name = "oxy-server"
3	+	version.workspace = true
4	+	edition.workspace = true
5	+	license.workspace = true
6	+	description = "Reference HTTP server for the Oxygen wire protocol"
7	+
8	+	[dependencies]
9	+	oxy-core = { path = "../oxy-core" }
10	+	axum = "0.8"
11	+	tokio = { version = "1", features = ["rt-multi-thread", "net", "macros", "sync"] }
12	+
13	+	[dev-dependencies]
14	+	tempfile = "3"

1	+	//! Reference HTTP server for the Oxygen wire protocol (spec §15.1, §16).
2	+	//!
3	+	//! Serves every repository directory under a root at `/{repo}/oxy/v1/...`.
4	+	//! TLS is expected to be terminated in front of this server for now; the
5	+	//! protocol's integrity does not depend on the channel (spec §15).
6	+	//!
7	+	//! Authentication: when bearer tokens are configured, `push` requires one;
8	+	//! reads are open. With no tokens configured everything is open (development).
9	+
10	+	use axum::body::Bytes;
11	+	use axum::extract::{Path, State};
12	+	use axum::http::{header, HeaderMap, StatusCode};
13	+	use axum::response::{IntoResponse, Response};
14	+	use axum::routing::{get, post};
15	+	use axum::Router;
16	+	use oxy_core::protocol::{wire, CONTENT_TYPE};
17	+	use oxy_core::repo::Repo;
18	+	use std::collections::HashMap;
19	+	use std::path::PathBuf;
20	+	use std::sync::{Arc, Mutex};
21	+
22	+	#[derive(Clone, Debug)]
23	+	pub struct ServerConfig {
24	+	/// Directory containing one repository per subdirectory.
25	+	pub root: PathBuf,
26	+	/// Bearer tokens allowed to push; empty means no authentication.
27	+	pub tokens: Vec<String>,
28	+	}
29	+
30	+	struct App {
31	+	cfg: ServerConfig,
32	+	/// One lock per repository so concurrent pushes serialize.
33	+	locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
34	+	}
35	+
36	+	type Shared = Arc<App>;
37	+
38	+	struct Failure(StatusCode, String);
39	+
40	+	impl IntoResponse for Failure {
41	+	fn into_response(self) -> Response {
42	+	(self.0, self.1).into_response()
43	+	}
44	+	}
45	+
46	+	impl From<oxy_core::repo::RepoError> for Failure {
47	+	fn from(e: oxy_core::repo::RepoError) -> Failure {
48	+	Failure(StatusCode::BAD_REQUEST, e.to_string())
49	+	}
50	+	}
51	+
52	+	fn ok(body: Vec<u8>) -> Response {
53	+	([(header::CONTENT_TYPE, CONTENT_TYPE)], body).into_response()
54	+	}
55	+
56	+	fn valid_repo_name(name: &str) -> bool {
57	+	!name.is_empty()
58	+	&& !name.starts_with('.')
59	+	&& name.chars().all(\|c\| c.is_ascii_alphanumeric() \|\| matches!(c, '.' \| '_' \| '-'))
60	+	}
61	+
62	+	impl App {
63	+	fn open(&self, name: &str) -> Result<Repo, Failure> {
64	+	if !valid_repo_name(name) {
65	+	return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
66	+	}
67	+	let dir = self.cfg.root.join(name);
68	+	if !dir.join(".oxy").is_dir() {
69	+	return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
70	+	}
71	+	Repo::open(&dir).map_err(\|e\| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
72	+	}
73	+
74	+	fn lock_for(&self, name: &str) -> Arc<Mutex<()>> {
75	+	self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()
76	+	}
77	+
78	+	fn authorize_push(&self, headers: &HeaderMap) -> Result<(), Failure> {
79	+	if self.cfg.tokens.is_empty() {
80	+	return Ok(());
81	+	}
82	+	let presented = headers
83	+	.get(header::AUTHORIZATION)
84	+	.and_then(\|v\| v.to_str().ok())
85	+	.and_then(\|v\| v.strip_prefix("Bearer "))
86	+	.map(str::trim);
87	+	match presented {
88	+	Some(t) if self.cfg.tokens.iter().any(\|x\| x == t) => Ok(()),
89	+	_ => Err(Failure(StatusCode::UNAUTHORIZED, "a valid bearer token is required to push".into())),
90	+	}
91	+	}
92	+	}
93	+
94	+	async fn blocking<T: Send + 'static>(f: impl FnOnce() -> Result<T, Failure> + Send + 'static) -> Result<T, Failure> {
95	+	tokio::task::spawn_blocking(f).await.map_err(\|e\| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
96	+	}
97	+
98	+	async fn capabilities(State(app): State<Shared>, Path(repo): Path<String>) -> Result<Response, Failure> {
99	+	blocking(move \|\| {
100	+	let r = app.open(&repo)?;
101	+	Ok(ok(wire::encode_capabilities(r.id()?)))
102	+	})
103	+	.await
104	+	}
105	+
106	+	async fn refs(State(app): State<Shared>, Path(repo): Path<String>) -> Result<Response, Failure> {
107	+	blocking(move \|\| {
108	+	let r = app.open(&repo)?;
109	+	Ok(ok(wire::encode_refs(r.id()?, &r.remote_refs()?)))
110	+	})
111	+	.await
112	+	}
113	+
114	+	async fn negotiate(State(app): State<Shared>, Path(repo): Path<String>, body: Bytes) -> Result<Response, Failure> {
115	+	blocking(move \|\| {
116	+	let r = app.open(&repo)?;
117	+	let (have, want) = wire::decode_negotiate(&body)?;
118	+	Ok(ok(wire::encode_hashes(&r.negotiate(&have, &want)?)))
119	+	})
120	+	.await
121	+	}
122	+
123	+	async fn fetch(State(app): State<Shared>, Path(repo): Path<String>, body: Bytes) -> Result<Response, Failure> {
124	+	blocking(move \|\| {
125	+	let r = app.open(&repo)?;
126	+	let changes = wire::decode_hashes(&body)?;
127	+	Ok(ok(wire::encode_bundle(&r.bundle_for(&changes)?)))
128	+	})
129	+	.await
130	+	}
131	+
132	+	async fn push(State(app): State<Shared>, Path(repo): Path<String>, headers: HeaderMap, body: Bytes) -> Result<Response, Failure> {
133	+	app.authorize_push(&headers)?;
134	+	blocking(move \|\| {
135	+	let r = app.open(&repo)?;
136	+	let (bundle, updates) = wire::decode_push(&body)?;
137	+	let lock = app.lock_for(&repo);
138	+	let _guard = lock.lock().unwrap();
139	+	Ok(ok(wire::encode_push_response(&r.receive_push(bundle, updates)?)))
140	+	})
141	+	.await
142	+	}
143	+
144	+	pub fn router(cfg: ServerConfig) -> Router {
145	+	let app: Shared = Arc::new(App { cfg, locks: Mutex::new(HashMap::new()) });
146	+	Router::new()
147	+	.route("/{repo}/oxy/v1/capabilities", get(capabilities))
148	+	.route("/{repo}/oxy/v1/refs", get(refs))
149	+	.route("/{repo}/oxy/v1/negotiate", post(negotiate))
150	+	.route("/{repo}/oxy/v1/fetch", post(fetch))
151	+	.route("/{repo}/oxy/v1/push", post(push))
152	+	.with_state(app)
153	+	}
154	+
155	+	/// Serve on an already-bound listener until the process ends.
156	+	pub async fn serve(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
157	+	listener.set_nonblocking(true)?;
158	+	let listener = tokio::net::TcpListener::from_std(listener)?;
159	+	axum::serve(listener, router(cfg)).await
160	+	}
161	+
162	+	/// Build a runtime and serve; for the CLI and for tests on a background thread.
163	+	pub fn run_blocking(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
164	+	let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build()?;
165	+	rt.block_on(serve(listener, cfg))
166	+	}

1	1		[workspace]
2		-	members = ["crates/oxy-core", "."]
	2	+	members = ["crates/oxy-core", "crates/oxy-server", "."]
3	3		resolver = "2"
4	4
5	5		[workspace.package]
		…	15 unchanged lines
21	21
22	22		[dependencies]
23	23		oxy-core = { path = "crates/oxy-core" }
	24	+	oxy-server = { path = "crates/oxy-server" }
24	25		clap = { version = "4", features = ["derive"] }
25	26		anyhow = "1"
26	27		libc = "0.2"
		…	3 unchanged lines

		…	60 unchanged lines
61	61		pub bookmark: String,
62	62		/// `remote.<name>.url`.
63	63		pub remotes: BTreeMap<String, String>,
	64	+	/// `remote.<name>.token`: bearer token for HTTP pushes.
	65	+	pub remote_tokens: BTreeMap<String, String>,
64	66		}
65	67
66	68		impl Config {
		…	6 unchanged lines
73	75		sign_changes: false,
74	76		bookmark: "main".into(),
75	77		remotes: BTreeMap::new(),
	78	+	remote_tokens: BTreeMap::new(),
76	79		}
77	80		}
78	81
		…	17 unchanged lines
96	99		}
97	100		}
98	101		"workspace.bookmark" => self.bookmark = value.into(),
99		-	k if k.starts_with("remote.") && k.ends_with(".url") => {
100		-	let name = &k["remote.".len()..k.len() - ".url".len()];
	102	+	k if k.starts_with("remote.") && (k.ends_with(".url") \|\| k.ends_with(".token")) => {
	103	+	let (name, is_url) = match k.strip_suffix(".url") {
	104	+	Some(n) => (&n["remote.".len()..], true),
	105	+	None => (&k["remote.".len()..k.len() - ".token".len()], false),
	106	+	};
101	107		if name.is_empty() \|\| name.contains('/') {
102	108		return fail(format!("invalid remote name in {key:?}"));
103	109		}
	110	+	let map = if is_url { &mut self.remotes } else { &mut self.remote_tokens };
104	111		if value.is_empty() {
105		-	self.remotes.remove(name);
	112	+	map.remove(name);
	113	+	if is_url {
	114	+	self.remote_tokens.remove(name);
	115	+	}
106	116		} else {
107		-	self.remotes.insert(name.to_string(), value.into());
	117	+	map.insert(name.to_string(), value.into());
108	118		}
109	119		}
110	120		_ => return fail(format!("unknown configuration key {key:?}")),
		…	26 unchanged lines
137	147		);
138	148		for (name, url) in &self.remotes {
139	149		text.push_str(&format!("remote.{name}.url = \"{url}\"\n"));
	150	+	if let Some(t) = self.remote_tokens.get(name) {
	151	+	text.push_str(&format!("remote.{name}.token = \"{t}\"\n"));
	152	+	}
140	153		}
141	154		write_atomic(path, text.as_bytes())
142	155		}
		…	620 unchanged lines
763	776		self.config.remotes.get(name).cloned().ok_or_else(\|\| RepoError::Msg(format!("no remote named {name:?}; add one with `oxy remote add`")))
764	777		}
765	778
	779	+	/// Token for a remote: `remote.<name>.token`, else `$OXY_TOKEN`.
	780	+	fn remote_token(&self, name: &str) -> Option<String> {
	781	+	self.config.remote_tokens.get(name).cloned().or_else(\|\| std::env::var("OXY_TOKEN").ok())
	782	+	}
	783	+
	784	+	fn open_named_remote(&self, name: &str) -> Result<Box<dyn crate::protocol::Remote>> {
	785	+	open_remote(&self.remote_url(name)?, self.remote_token(name))
	786	+	}
	787	+
766	788		fn serial_path(&self, remote: &str, bookmark: &str) -> PathBuf {
767	789		self.oxy.join("remotes").join(remote).join(format!("{bookmark}.serial"))
768	790		}
		…	22 unchanged lines
791	813
792	814		/// Fetch every bookmark of a remote into `remotes/<name>/bookmarks/*` (spec §10.2).
793	815		pub fn fetch(&self, remote_name: &str) -> Result<Vec<(String, Vec<ChangeHash>, usize)>> {
794		-	let remote = open_remote(&self.remote_url(remote_name)?)?;
	816	+	let remote = self.open_named_remote(remote_name)?;
795	817		if remote.repo_id()? != self.id()? {
796	818		return fail(format!("remote {remote_name:?} is a different repository (repository IDs differ)"));
797	819		}
		…	80 unchanged lines
878	900		/// Push the current bookmark (spec §16.3), sealing this author's unsigned
879	901		/// changes and attesting the reference when a key is available.
880	902		pub fn push(&self, remote_name: &str, force: bool) -> Result<Vec<RefResult>> {
881		-	let remote = open_remote(&self.remote_url(remote_name)?)?;
	903	+	let remote = self.open_named_remote(remote_name)?;
882	904		if remote.repo_id()? != self.id()? {
883	905		return fail(format!("remote {remote_name:?} is a different repository (repository IDs differ)"));
884	906		}
		…	74 unchanged lines
959	981		/// Clone a remote into `dest`: same repository ID, `origin` remote,
960	982		/// bookmark set to the remote's, working copy checked out.
961	983		pub fn clone(url: &str, dest: &Path) -> Result<Repo> {
962		-	let remote = open_remote(url)?;
	984	+	let remote = open_remote(url, std::env::var("OXY_TOKEN").ok())?;
963	985		let id = remote.repo_id()?;
964	986		let mut repo = Repo::init_with_id(dest, id)?;
965	987		repo.config.remotes.insert("origin".into(), url.to_string());
		…	1552 unchanged lines

		…	58 unchanged lines
59	59		},
60	60		/// Report which conflicts of the working copy's base are resolved by the current files
61	61		Resolve,
	62	+	/// Serve repositories under ROOT over HTTP (spec §16); TLS is terminated in front
	63	+	Serve {
	64	+	/// Directory with one repository per subdirectory
	65	+	#[arg(long, default_value = ".")]
	66	+	root: PathBuf,
	67	+	#[arg(long, default_value = "127.0.0.1:8890")]
	68	+	bind: String,
	69	+	/// Bearer token allowed to push (repeatable); none means open
	70	+	#[arg(long = "token")]
	71	+	tokens: Vec<String>,
	72	+	},
62	73		/// Undo the last operation
63	74		Undo,
64	75		/// Manage the signing identity
		…	3 unchanged lines
68	79		},
69	80		/// Verify the signatures on a change (default: every change in the log)
70	81		Verify { change: Option<String> },
71		-	/// Set a repository configuration key (author.name, author.email, author.key, core.sign, workspace.bookmark)
	82	+	/// Set a repository configuration key (author.*, core.sign, workspace.bookmark, remote.<name>.url\|token)
72	83		Config { key: String, value: String },
73	84		/// List operations
74	85		#[command(name = "op")]
		…	317 unchanged lines
392	403		Ok(2)
393	404		}
394	405		}
	406	+	Cmd::Serve { root, bind, tokens } => {
	407	+	let listener = std::net::TcpListener::bind(&bind).with_context(\|\| format!("binding {bind}"))?;
	408	+	let root = root.canonicalize()?;
	409	+	println!("Serving repositories under {} on http://{bind}/<repo>/oxy/v1/ ({})", root.display(), if tokens.is_empty() { "open: no push tokens configured" } else { "push requires a token" });
	410	+	oxy_server::run_blocking(listener, oxy_server::ServerConfig { root, tokens })?;
	411	+	Ok(0)
	412	+	}
395	413		Cmd::Undo => {
396	414		let repo = open()?;
397	415		let op = repo.undo()?;
		…	72 unchanged lines

1	+	//! End-to-end: a real server on a loopback port, clients over HTTP.
2	+
3	+	use oxy_core::repo::{Repo, SignatureStatus};
4	+	use std::fs;
5	+	use std::net::TcpListener;
6	+	use std::path::Path;
7	+
8	+	fn write(root: &Path, rel: &str, content: &[u8]) {
9	+	let p = root.join(rel);
10	+	fs::create_dir_all(p.parent().unwrap()).unwrap();
11	+	fs::write(p, content).unwrap();
12	+	}
13	+
14	+	fn start(root: &Path, tokens: Vec<String>) -> String {
15	+	let listener = TcpListener::bind("127.0.0.1:0").unwrap();
16	+	let port = listener.local_addr().unwrap().port();
17	+	let cfg = oxy_server::ServerConfig { root: root.to_path_buf(), tokens };
18	+	std::thread::spawn(move \|\| oxy_server::run_blocking(listener, cfg).unwrap());
19	+	format!("http://127.0.0.1:{port}")
20	+	}
21	+
22	+	#[test]
23	+	fn clone_push_pull_over_http() {
24	+	let dir = tempfile::tempdir().unwrap();
25	+	let root = dir.path();
26	+	let served = root.join("served");
27	+	Repo::init(&served.join("origin")).unwrap();
28	+	let base = start(&served, vec![]);
29	+	let url = format!("{base}/origin");
30	+
31	+	let mut a = Repo::clone(&url, &root.join("a")).unwrap().with_keys_dir(root.join("keys"));
32	+	let (key, _) = a.create_identity("Ada", "[email protected]").unwrap();
33	+	write(&root.join("a"), "f.txt", b"hello\n");
34	+	let h1 = a.commit("hello", &[]).unwrap();
35	+	let r = a.push("origin", false).unwrap();
36	+	assert!(r[0].ok, "{}", r[0].reason);
37	+
38	+	let origin = Repo::open(&served.join("origin")).unwrap();
39	+	assert_eq!(origin.bookmark_heads().unwrap(), vec![h1]);
40	+	assert!(matches!(origin.authenticate_change(&h1).unwrap(), SignatureStatus::Valid { key: k, sealed: true, .. } if k == key));
41	+
42	+	let b = Repo::clone(&url, &root.join("b")).unwrap();
43	+	assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"hello\n");
44	+	write(&root.join("b"), "g.txt", b"g\n");
45	+	let h2 = b.commit("g", &[]).unwrap();
46	+	assert!(b.push("origin", false).unwrap()[0].ok);
47	+
48	+	let pr = a.pull("origin").unwrap();
49	+	assert!(pr.changed && pr.checked_out);
50	+	assert_eq!(fs::read(root.join("a/g.txt")).unwrap(), b"g\n");
51	+	let mut heads = a.bookmark_heads().unwrap();
52	+	heads.sort();
53	+	let mut want = vec![h1, h2];
54	+	want.sort();
55	+	assert_eq!(heads, want);
56	+
57	+	// Unknown repositories and names the server refuses are surfaced as errors.
58	+	let err = Repo::clone(&format!("{base}/nope"), &root.join("c")).unwrap_err().to_string();
59	+	assert!(err.contains("404"), "{err}");
60	+	fs::create_dir_all(served.join(".hidden/.oxy")).unwrap();
61	+	assert!(Repo::clone(&format!("{base}/.hidden"), &root.join("d")).unwrap_err().to_string().contains("404"));
62	+	}
63	+
64	+	#[test]
65	+	fn push_requires_a_token_when_configured() {
66	+	let dir = tempfile::tempdir().unwrap();
67	+	let root = dir.path();
68	+	let served = root.join("served");
69	+	Repo::init(&served.join("origin")).unwrap();
70	+	let base = start(&served, vec!["s3cret".into()]);
71	+	let url = format!("{base}/origin");
72	+
73	+	// Reads are open.
74	+	let mut a = Repo::clone(&url, &root.join("a")).unwrap();
75	+	write(&root.join("a"), "f.txt", b"x\n");
76	+	a.commit("x", &[]).unwrap();
77	+	let err = a.push("origin", false).unwrap_err().to_string();
78	+	assert!(err.contains("401"), "{err}");
79	+
80	+	a.config.set("remote.origin.token", "wrong").unwrap();
81	+	a.save_config().unwrap();
82	+	assert!(a.push("origin", false).unwrap_err().to_string().contains("401"));
83	+
84	+	a.config.set("remote.origin.token", "s3cret").unwrap();
85	+	a.save_config().unwrap();
86	+	assert!(a.push("origin", false).unwrap()[0].ok);
87	+	let origin = Repo::open(&served.join("origin")).unwrap();
88	+	assert_eq!(origin.bookmark_heads().unwrap().len(), 1);
89	+	}

		…	13 unchanged lines
14	14		ignore = "0.4"
15	15		ml-dsa = "0.1"
16	16		ed25519-dalek = "3"
	17	+	ureq = "2"
17	18
18	19		[dev-dependencies]
19	20		tempfile = "3"

1	1		//! Transport-agnostic protocol (spec §16): what a client asks of a remote and
2	2		//! what a server does with a push. `LocalRemote` serves another repository
3		-	//! directory in-process; an HTTPS transport implements the same trait and the
4		-	//! server side reuses `Repo::receive_push` and friends unchanged.
	3	+	//! directory in-process, `HttpRemote` speaks to an `oxy-server` over HTTP(S),
	4	+	//! and the server side reuses `Repo::receive_push` and friends unchanged.
	5	+	//!
	6	+	//! Wire messages are deterministic CBOR (`wire` module); over HTTP each is a
	7	+	//! request or response body with content type `application/oxy-cbor`.
5	8
6	9		use crate::attest::Attest;
	10	+	use crate::cbor::{self, Value};
7	11		use crate::hash::{ChangeHash, ObjectRef, ObjectType};
8	12		use crate::identity::Signature;
9		-	use crate::objects::Change;
	13	+	use crate::objects::{self as o, Change, ObjectError};
10	14		use crate::repo::{Repo, RepoError, Result};
11	15		use std::path::Path;
12	16
	17	+	pub const CONTENT_TYPE: &str = "application/oxy-cbor";
	18	+	pub const PROTOCOL_VERSION: u64 = 1;
	19	+
13	20		/// An attestation together with the signatures over its ObjectRef.
14	21		#[derive(Clone, Debug, PartialEq, Eq)]
15	22		pub struct SignedAttest {
		…	53 unchanged lines
69	76		fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>>;
70	77		}
71	78
	79	+	// ---------------------------------------------------------------- local
	80	+
72	81		/// Another repository on this machine, served in-process.
73	82		pub struct LocalRemote {
74	83		pub repo: Repo,
		…	23 unchanged lines
98	107		}
99	108		}
100	109
101		-	/// Open a remote by URL. Only local paths (optionally `file://`) exist so far.
102		-	pub fn open_remote(url: &str) -> Result<Box<dyn Remote>> {
	110	+	// ---------------------------------------------------------------- http
	111	+
	112	+	/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
	113	+	pub struct HttpRemote {
	114	+	base: String,
	115	+	token: Option<String>,
	116	+	agent: ureq::Agent,
	117	+	}
	118	+
	119	+	impl HttpRemote {
	120	+	pub fn new(base: &str, token: Option<String>) -> HttpRemote {
	121	+	HttpRemote { base: base.trim_end_matches('/').to_string(), token, agent: ureq::Agent::new() }
	122	+	}
	123	+
	124	+	fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
	125	+	let url = format!("{}/oxy/v1/{endpoint}", self.base);
	126	+	let mut req = if post { self.agent.post(&url) } else { self.agent.get(&url) };
	127	+	req = req.set("Accept", CONTENT_TYPE);
	128	+	if let Some(t) = &self.token {
	129	+	req = req.set("Authorization", &format!("Bearer {t}"));
	130	+	}
	131	+	let resp = match body {
	132	+	Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
	133	+	None => req.call(),
	134	+	};
	135	+	match resp {
	136	+	Ok(r) => {
	137	+	let mut out = Vec::new();
	138	+	r.into_reader().read_to_end(&mut out).map_err(\|e\| RepoError::Msg(format!("{url}: {e}")))?;
	139	+	Ok(out)
	140	+	}
	141	+	Err(ureq::Error::Status(code, r)) => {
	142	+	let text = r.into_string().unwrap_or_default();
	143	+	Err(RepoError::Msg(format!("{url}: HTTP {code}: {}", text.trim())))
	144	+	}
	145	+	Err(e) => Err(RepoError::Msg(format!("{url}: {e}"))),
	146	+	}
	147	+	}
	148	+	}
	149	+
	150	+	impl Remote for HttpRemote {
	151	+	fn repo_id(&self) -> Result<[u8; 16]> {
	152	+	wire::decode_capabilities(&self.call(false, "capabilities", None)?)
	153	+	}
	154	+	fn refs(&self) -> Result<Vec<RemoteRef>> {
	155	+	Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
	156	+	}
	157	+	fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
	158	+	wire::decode_hashes(&self.call(true, "negotiate", Some(&wire::encode_negotiate(have, want)))?)
	159	+	}
	160	+	fn fetch(&self, changes: &[ChangeHash]) -> Result<Bundle> {
	161	+	wire::decode_bundle(&self.call(true, "fetch", Some(&wire::encode_hashes(changes)))?)
	162	+	}
	163	+	fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>) -> Result<Vec<RefResult>> {
	164	+	wire::decode_push_response(&self.call(true, "push", Some(&wire::encode_push(&bundle, &updates)))?)
	165	+	}
	166	+	}
	167	+
	168	+	/// Open a remote by URL: a local path (optionally `file://`), or `http(s)://`.
	169	+	pub fn open_remote(url: &str, token: Option<String>) -> Result<Box<dyn Remote>> {
	170	+	if url.starts_with("http://") \|\| url.starts_with("https://") {
	171	+	return Ok(Box::new(HttpRemote::new(url, token)));
	172	+	}
103	173		let path = url.strip_prefix("file://").unwrap_or(url);
104	174		if path.contains("://") {
105		-	return Err(RepoError::Msg(format!("unsupported remote URL {url:?}: only local paths are implemented")));
	175	+	return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
106	176		}
107	177		Ok(Box::new(LocalRemote::open(Path::new(path))?))
108	178		}
	179	+
	180	+	// ---------------------------------------------------------------- wire
	181	+
	182	+	/// CBOR encodings of the protocol messages (spec §16.1).
	183	+	pub mod wire {
	184	+	use super::*;
	185	+
	186	+	fn schema(msg: &str) -> RepoError {
	187	+	RepoError::Object(ObjectError::Schema(msg.into()))
	188	+	}
	189	+	fn sigs_value(sigs: &[Signature]) -> Value {
	190	+	Value::Array(sigs.iter().map(\|s\| s.to_value()).collect())
	191	+	}
	192	+	fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
	193	+	Ok(o::array(v, "signatures")?.iter().map(Signature::from_value).collect::<std::result::Result<Vec<_>, _>>()?)
	194	+	}
	195	+	fn hashes_value(hs: &[ObjectRef]) -> Value {
	196	+	Value::Array(hs.iter().map(\|h\| Value::Bytes(h.0.to_vec())).collect())
	197	+	}
	198	+	fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
	199	+	Ok(o::array(v, "hashes")?.iter().map(\|h\| o::oref(h, "hash")).collect::<std::result::Result<Vec<_>, _>>()?)
	200	+	}
	201	+	fn attest_value(sa: &SignedAttest) -> Value {
	202	+	Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
	203	+	}
	204	+	fn attest_from(v: &Value) -> Result<SignedAttest> {
	205	+	let a = o::array(v, "signed attest")?;
	206	+	if a.len() != 2 {
	207	+	return Err(schema("signed attest: expected [attest, sigs]"));
	208	+	}
	209	+	Ok(SignedAttest { attest: Attest::decode(o::bytes(&a[0], "attest")?)?, sigs: sigs_from(&a[1])? })
	210	+	}
	211	+	fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
	212	+	sa.as_ref().map(attest_value).unwrap_or(Value::Null)
	213	+	}
	214	+	fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
	215	+	if v.is_null() { Ok(None) } else { attest_from(v).map(Some) }
	216	+	}
	217	+	fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
	218	+	Ok(o::map(v, what)?)
	219	+	}
	220	+	fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
	221	+	Ok(o::field(m, k, what)?)
	222	+	}
	223	+
	224	+	pub fn bundle_value(b: &Bundle) -> Value {
	225	+	Value::map([
	226	+	(
	227	+	0,
	228	+	Value::Array(b.changes.iter().map(\|(c, s)\| Value::Array(vec![Value::Bytes(c.encode()), sigs_value(s)])).collect()),
	229	+	),
	230	+	(1, Value::Array(b.attests.iter().map(attest_value).collect())),
	231	+	(
	232	+	2,
	233	+	Value::Array(
	234	+	b.objects.iter().map(\|(t, body)\| Value::Array(vec![Value::Text(t.name().into()), Value::Bytes(body.clone())])).collect(),
	235	+	),
	236	+	),
	237	+	])
	238	+	}
	239	+
	240	+	pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
	241	+	let m = map(v, "bundle")?;
	242	+	let mut b = Bundle::default();
	243	+	for c in o::array(field(m, 0, "changes")?, "changes")? {
	244	+	let pair = o::array(c, "change entry")?;
	245	+	if pair.len() != 2 {
	246	+	return Err(schema("change entry: expected [change, sigs]"));
	247	+	}
	248	+	b.changes.push((Change::decode(o::bytes(&pair[0], "change")?)?, sigs_from(&pair[1])?));
	249	+	}
	250	+	for a in o::array(field(m, 1, "attests")?, "attests")? {
	251	+	b.attests.push(attest_from(a)?);
	252	+	}
	253	+	for ob in o::array(field(m, 2, "objects")?, "objects")? {
	254	+	let pair = o::array(ob, "object")?;
	255	+	if pair.len() != 2 {
	256	+	return Err(schema("object: expected [type, body]"));
	257	+	}
	258	+	let ty = ObjectType::from_name(o::text(&pair[0], "object type")?).ok_or_else(\|\| schema("unknown object type"))?;
	259	+	b.objects.push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
	260	+	}
	261	+	Ok(b)
	262	+	}
	263	+
	264	+	pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
	265	+	cbor::encode(&bundle_value(b))
	266	+	}
	267	+	pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
	268	+	bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
	269	+	}
	270	+
	271	+	pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
	272	+	cbor::encode(&Value::map([
	273	+	(0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
	274	+	(1, Value::Bytes(id.to_vec())),
	275	+	(2, Value::Array(vec![Value::Text("seal".into())])),
	276	+	]))
	277	+	}
	278	+	pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
	279	+	let v = cbor::decode(bytes).map_err(ObjectError::from)?;
	280	+	let m = map(&v, "capabilities")?;
	281	+	let versions = o::array(field(m, 0, "versions")?, "versions")?;
	282	+	if !versions.iter().any(\|x\| x.as_uint() == Some(PROTOCOL_VERSION)) {
	283	+	return Err(RepoError::Msg("remote does not speak protocol version 1".into()));
	284	+	}
	285	+	o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(\|_\| schema("repo id: expected 16 bytes"))
	286	+	}
	287	+
	288	+	pub fn encode_refs(id: [u8; 16], refs: &[RemoteRef]) -> Vec<u8> {
	289	+	cbor::encode(&Value::map([
	290	+	(
	291	+	0,
	292	+	Value::Array(
	293	+	refs.iter()
	294	+	.map(\|r\| {
	295	+	Value::map([
	296	+	(0, Value::Text(r.name.clone())),
	297	+	(1, Value::Bytes(r.state.0.to_vec())),
	298	+	(2, hashes_value(&r.heads)),
	299	+	(3, opt_attest_value(&r.attestation)),
	300	+	(4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
	301	+	])
	302	+	})
	303	+	.collect(),
	304	+	),
	305	+	),
	306	+	(1, Value::Bytes(id.to_vec())),
	307	+	]))
	308	+	}
	309	+	pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], Vec<RemoteRef>)> {
	310	+	let v = cbor::decode(bytes).map_err(ObjectError::from)?;
	311	+	let m = map(&v, "refs")?;
	312	+	let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(\|_\| schema("repo id: expected 16 bytes"))?;
	313	+	let mut refs = Vec::new();
	314	+	for r in o::array(field(m, 0, "refs")?, "refs")? {
	315	+	let rm = map(r, "ref")?;
	316	+	let ls = field(rm, 4, "last serial")?;
	317	+	refs.push(RemoteRef {
	318	+	name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
	319	+	state: o::oref(field(rm, 1, "state")?, "state")?,
	320	+	heads: hashes_from(field(rm, 2, "heads")?)?,
	321	+	attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
	322	+	last_serial: if ls.is_null() { None } else { Some(o::uint(ls, "last serial")?) },
	323	+	});
	324	+	}
	325	+	Ok((id, refs))
	326	+	}
	327	+
	328	+	pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
	329	+	cbor::encode(&Value::map([(0, hashes_value(have)), (1, hashes_value(want))]))
	330	+	}
	331	+	pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
	332	+	let v = cbor::decode(bytes).map_err(ObjectError::from)?;
	333	+	let m = map(&v, "negotiate")?;
	334	+	Ok((hashes_from(field(m, 0, "have")?)?, hashes_from(field(m, 1, "want")?)?))
	335	+	}
	336	+
	337	+	pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
	338	+	cbor::encode(&Value::map([(0, hashes_value(hs))]))
	339	+	}
	340	+	pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
	341	+	let v = cbor::decode(bytes).map_err(ObjectError::from)?;
	342	+	hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
	343	+	}
	344	+
	345	+	pub fn encode_push(bundle: &Bundle, updates: &[RefUpdate]) -> Vec<u8> {
	346	+	cbor::encode(&Value::map([
	347	+	(0, bundle_value(bundle)),
	348	+	(
	349	+	1,
	350	+	Value::Array(
	351	+	updates
	352	+	.iter()
	353	+	.map(\|u\| {
	354	+	Value::map([
	355	+	(0, Value::Text(u.name.clone())),
	356	+	(1, Value::Bytes(u.old_state.0.to_vec())),
	357	+	(2, Value::Bytes(u.new_state.0.to_vec())),
	358	+	(3, Value::Bool(u.force)),
	359	+	(4, opt_attest_value(&u.attestation)),
	360	+	])
	361	+	})
	362	+	.collect(),
	363	+	),
	364	+	),
	365	+	]))
	366	+	}
	367	+	pub fn decode_push(bytes: &[u8]) -> Result<(Bundle, Vec<RefUpdate>)> {
	368	+	let v = cbor::decode(bytes).map_err(ObjectError::from)?;
	369	+	let m = map(&v, "push")?;
	370	+	let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
	371	+	let mut updates = Vec::new();
	372	+	for u in o::array(field(m, 1, "updates")?, "updates")? {
	373	+	let um = map(u, "update")?;
	374	+	updates.push(RefUpdate {
	375	+	name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
	376	+	old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
	377	+	new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
	378	+	force: field(um, 3, "force")?.as_bool().ok_or_else(\|\| schema("force: expected bool"))?,
	379	+	attestation: opt_attest_from(field(um, 4, "attestation")?)?,
	380	+	});
	381	+	}
	382	+	Ok((bundle, updates))
	383	+	}
	384	+
	385	+	pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
	386	+	cbor::encode(&Value::map([(
	387	+	0,
	388	+	Value::Array(
	389	+	results
	390	+	.iter()
	391	+	.map(\|r\| Value::map([(0, Value::Text(r.name.clone())), (1, Value::Bool(r.ok)), (2, Value::Text(r.reason.clone()))]))
	392	+	.collect(),
	393	+	),
	394	+	)]))
	395	+	}
	396	+	pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
	397	+	let v = cbor::decode(bytes).map_err(ObjectError::from)?;
	398	+	let m = map(&v, "push response")?;
	399	+	let mut out = Vec::new();
	400	+	for r in o::array(field(m, 0, "results")?, "results")? {
	401	+	let rm = map(r, "result")?;
	402	+	out.push(RefResult {
	403	+	name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
	404	+	ok: field(rm, 1, "ok")?.as_bool().ok_or_else(\|\| schema("ok: expected bool"))?,
	405	+	reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
	406	+	});
	407	+	}
	408	+	Ok(out)
	409	+	}
	410	+	}
	411	+
	412	+	#[cfg(test)]
	413	+	mod tests {
	414	+	use super::*;
	415	+	use crate::attest::AttestBody;
	416	+	use crate::objects::Hlc;
	417	+
	418	+	#[test]
	419	+	fn wire_round_trips() {
	420	+	let c = crate::objects::tests::sample_change();
	421	+	let sk = crate::identity::SecretKey { ed_seed: [1; 32], ml_seed: [2; 32] };
	422	+	let id = sk.identity("a", "b", 0);
	423	+	let key = id.hash();
	424	+	let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
	425	+	let seal = Attest::seal([3; 16], vec![c.hash()]);
	426	+	let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
	427	+	let bundle = Bundle {
	428	+	changes: vec![(c.clone(), vec![sig.clone()])],
	429	+	attests: vec![SignedAttest { attest: seal.clone(), sigs: vec![seal_sig] }],
	430	+	objects: vec![(ObjectType::Identity, id.encode()), (ObjectType::Chunk, b"xyz".to_vec())],
	431	+	};
	432	+	let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
	433	+	assert_eq!(d.changes, bundle.changes);
	434	+	assert_eq!(d.attests, bundle.attests);
	435	+	assert_eq!(d.objects, bundle.objects);
	436	+
	437	+	let att = Attest { repo: [3; 16], hlc: Hlc::default(), body: AttestBody::Ref { name: "main".into(), old_state: ObjectRef([1; 32]), new_state: ObjectRef([2; 32]), serial: 4 } };
	438	+	let refs = vec![
	439	+	RemoteRef { name: "main".into(), state: ObjectRef([2; 32]), heads: vec![c.hash()], attestation: Some(SignedAttest { attest: att.clone(), sigs: vec![sig.clone()] }), last_serial: Some(4) },
	440	+	RemoteRef { name: "dev".into(), state: ObjectRef([5; 32]), heads: vec![], attestation: None, last_serial: None },
	441	+	];
	442	+	let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &refs)).unwrap();
	443	+	assert_eq!(id_back, [3; 16]);
	444	+	assert_eq!(back.len(), 2);
	445	+	assert_eq!(back[0].attestation.as_ref().unwrap().attest, att);
	446	+	assert_eq!(back[1].last_serial, None);
	447	+
	448	+	let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(&[ObjectRef([1; 32])], &[ObjectRef([2; 32])])).unwrap();
	449	+	assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
	450	+	assert_eq!(wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(), vec![c.hash()]);
	451	+
	452	+	let up = RefUpdate { name: "main".into(), old_state: ObjectRef([1; 32]), new_state: ObjectRef([2; 32]), force: true, attestation: Some(SignedAttest { attest: att, sigs: vec![sig] }) };
	453	+	let (b2, ups) = wire::decode_push(&wire::encode_push(&bundle, std::slice::from_ref(&up))).unwrap();
	454	+	assert_eq!(b2.changes.len(), 1);
	455	+	assert_eq!(ups[0].name, "main");
	456	+	assert!(ups[0].force);
	457	+	assert_eq!(ups[0].attestation, up.attestation);
	458	+	let res = vec![RefResult { name: "main".into(), ok: false, reason: "stale".into() }];
	459	+	assert_eq!(wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(), res);
	460	+	assert_eq!(wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(), [7; 16]);
	461	+	}
	462	+	}

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

		…	2 unchanged lines
3	3		version = 4
4	4
5	5		[[package]]
	6	+	name = "adler2"
	7	+	version = "2.0.1"
	8	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	9	+	checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
	10	+
	11	+	[[package]]
6	12		name = "aho-corasick"
7	13		version = "1.1.5"
8	14		source = "registry+https://github.com/rust-lang/crates.io-index"
		…	38 unchanged lines
47	53		source = "registry+https://github.com/rust-lang/crates.io-index"
48	54		checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
49	55		dependencies = [
50		-	"windows-sys",
	56	+	"windows-sys 0.61.2",
51	57		]
52	58
53	59		[[package]]
		…	4 unchanged lines
58	64		dependencies = [
59	65		"anstyle",
60	66		"once_cell_polyfill",
61		-	"windows-sys",
	67	+	"windows-sys 0.61.2",
62	68		]
63	69
64	70		[[package]]
		…	9 unchanged lines
74	80		checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
75	81
76	82		[[package]]
	83	+	name = "atomic-waker"
	84	+	version = "1.1.2"
	85	+	source = "registry+https://github.com/rust-lang/crates.io-index"
	86	+	checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
	87	+

HTTP transport: CBOR wire messages, HttpRemote client, oxy-server crate with bearer-token pushes, oxy serve 750251b21a96

Keyboard