crates/oxy-server/src/lib.rs at 8c30d814f2cc · andymccall/oxygen

# 8c30d814f2cc oxygen / crates / oxy-server / src / lib.rs

14.3 KiB · Rust · last changed A bare repository's default bookmark is its host's to protect 1 month ago source blame history raw

//! Reference HTTP server for the Oxygen wire protocol (spec §15.1, §16).
//!
//! Serves every repository directory under a root at `/{repo}/oxy/v1/...`,
//! over TLS when a certificate and key are given, else plain HTTP for use
//! behind a TLS-terminating proxy or on a loopback interface (spec §15.1).
//!
//! Authentication: when bearer tokens are configured, `push` requires one;
//! reads are open. With no tokens configured everything is open (development).
//!
//! Hosts embed the protocol by implementing [`Backend`] (name resolution
//! and authorization) and routing to [`Server::handle`]; `router` is the
//! standalone server built on [`DirBackend`].

use axum::body::Bytes;
use axum::extract::{Path, State};
use axum::http::{header, HeaderMap, Method, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::any;
use axum::Router;
use oxy_core::protocol::{wire, CONTENT_TYPE};
use oxy_core::repo::Repo;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

#[derive(Clone, Debug)]
pub struct ServerConfig {
    /// Directory containing one repository per subdirectory.
    pub root: PathBuf,
    /// Bearer tokens allowed to push; empty means no authentication.
    pub tokens: Vec<String>,
    /// Serve HTTPS with this certificate chain and private key (PEM files).
    pub tls: Option<TlsFiles>,
}

#[derive(Clone, Debug)]
pub struct TlsFiles {
    pub cert: PathBuf,
    pub key: PathBuf,
}

/// Which kind of access a request needs (spec §15.1: reads may be open;
/// a push must be authenticated when any credential is configured).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Access {
    Read,
    Push,
}

/// What a host provides to the protocol server: how a repository name maps
/// to a repository on disk and who may do what. `oxy serve` uses
/// [`DirBackend`]; a hub supplies its own with accounts and roles.
pub trait Backend: Send + Sync + 'static {
    /// Open the repository called `name`, or fail with the status to
    /// answer. An unknown name is `404`; a name the caller may not see
    /// SHOULD also be `404`, so private repositories are not enumerable.
    fn open(&self, name: &str) -> Result<Repo, Failure>;

    /// Decide whether this request may perform `access` on `name`.
    fn authorize(&self, name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure>;

    /// Called after a push was processed, with its per-reference results
    /// (accepted or refused), the states each accepted reference moved
    /// between, and the changes stored, for logs and cache invalidation.
    fn after_push(&self, _name: &str, _report: &oxy_core::repo::PushReport, _headers: &HeaderMap) {}

    /// The bookmark of `name` that pushes may never delete (its default),
    /// when the host keeps one apart from the repository's own config.
    fn default_bookmark(&self, _name: &str) -> Option<String> {
        None
    }
}

/// A protocol server over any [`Backend`], serialising the pushes to each
/// repository. Hosts route requests to [`Server::handle`].
pub struct Server {
    backend: Box<dyn Backend>,
    locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
}

/// An error response: a status and a plain-text body.
#[derive(Debug)]
pub struct Failure(pub StatusCode, pub String);

impl IntoResponse for Failure {
    fn into_response(self) -> Response {
        (self.0, self.1).into_response()
    }
}

impl From<oxy_core::repo::RepoError> for Failure {
    fn from(e: oxy_core::repo::RepoError) -> Failure {
        Failure(StatusCode::BAD_REQUEST, e.to_string())
    }
}

/// The wire protocol's endpoints (spec §16.1), named as in the URL.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Endpoint {
    Capabilities,
    Refs,
    Negotiate,
    Fetch,
    Objects,
    Checkpoint,
    Shallow,
    Push,
}

impl Endpoint {
    pub fn parse(name: &str) -> Option<Endpoint> {
        Some(match name {
            "capabilities" => Endpoint::Capabilities,
            "refs" => Endpoint::Refs,
            "negotiate" => Endpoint::Negotiate,
            "fetch" => Endpoint::Fetch,
            "objects" => Endpoint::Objects,
            "checkpoint" => Endpoint::Checkpoint,
            "shallow" => Endpoint::Shallow,
            "push" => Endpoint::Push,
            _ => return None,
        })
    }

    pub fn name(self) -> &'static str {
        match self {
            Endpoint::Capabilities => "capabilities",
            Endpoint::Refs => "refs",
            Endpoint::Negotiate => "negotiate",
            Endpoint::Fetch => "fetch",
            Endpoint::Objects => "objects",
            Endpoint::Checkpoint => "checkpoint",
            Endpoint::Shallow => "shallow",
            Endpoint::Push => "push",
        }
    }

    /// The HTTP method the endpoint answers to.
    pub fn method(self) -> Method {
        match self {
            Endpoint::Capabilities | Endpoint::Refs => Method::GET,
            _ => Method::POST,
        }
    }

    fn access(self) -> Access {
        match self {
            Endpoint::Push => Access::Push,
            _ => Access::Read,
        }
    }
}

fn ok(body: Vec<u8>) -> Response {
    ([(header::CONTENT_TYPE, CONTENT_TYPE)], body).into_response()
}

pub fn valid_repo_name(name: &str) -> bool {
    !name.is_empty()
        && !name.starts_with('.')
        && name
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
}

impl Server {
    pub fn new(backend: impl Backend) -> Arc<Server> {
        Arc::new(Server {
            backend: Box::new(backend),
            locks: Mutex::new(HashMap::new()),
        })
    }

    pub fn backend(&self) -> &dyn Backend {
        self.backend.as_ref()
    }

    fn lock_for(&self, name: &str) -> Arc<Mutex<()>> {
        self.locks
            .lock()
            .unwrap()
            .entry(name.to_string())
            .or_default()
            .clone()
    }

    /// Answer one protocol request for repository `name`. `method` is
    /// checked against the endpoint; the body is the raw CBOR request.
    pub async fn handle(
        self: &Arc<Self>,
        name: String,
        endpoint: Endpoint,
        method: &Method,
        headers: &HeaderMap,
        body: Bytes,
    ) -> Response {
        if *method != endpoint.method() {
            return Failure(
                StatusCode::METHOD_NOT_ALLOWED,
                format!("{} is {}", endpoint.name(), endpoint.method()),
            )
            .into_response();
        }
        if let Err(f) = self.backend.authorize(&name, endpoint.access(), headers) {
            return f.into_response();
        }
        let claimed = headers
            .get(oxy_core::protocol::REPO_ID_HEADER)
            .and_then(|v| v.to_str().ok())
            .and_then(hex_id);
        let server = Arc::clone(self);
        let request_headers = headers.clone();
        let result = tokio::task::spawn_blocking(move || {
            server.respond(&name, endpoint, body, claimed, &request_headers)
        })
        .await;
        match result {
            Ok(Ok(r)) => r,
            Ok(Err(f)) => f.into_response(),
            Err(e) => Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
        }
    }

    fn respond(
        &self,
        name: &str,
        endpoint: Endpoint,
        body: Bytes,
        claimed: Option<[u8; 16]>,
        headers: &HeaderMap,
    ) -> Result<Response, Failure> {
        let r = self.backend.open(name)?;
        Ok(match endpoint {
            Endpoint::Capabilities => ok(wire::encode_capabilities(r.id()?)),
            Endpoint::Refs => ok(wire::encode_refs(r.id()?, &r.remote_refs()?)),
            Endpoint::Negotiate => {
                let (have, want) = wire::decode_negotiate(&body)?;
                ok(wire::encode_hashes(&r.negotiate(&have, &want)?))
            }
            Endpoint::Fetch => {
                let (changes, max_blob) = wire::decode_fetch(&body)?;
                ok(wire::encode_bundle(
                    &r.bundle_for_filtered(&changes, max_blob)?,
                ))
            }
            Endpoint::Objects => {
                let refs = wire::decode_hashes(&body)?;
                let bundle = oxy_core::protocol::Bundle {
                    objects: r.objects_for(&refs)?,
                    ..Default::default()
                };
                ok(wire::encode_bundle(&bundle))
            }
            Endpoint::Checkpoint => {
                let heads = wire::decode_hashes(&body)?;
                let lock = self.lock_for(name);
                let _guard = lock.lock().unwrap();
                ok(wire::encode_bundle(&r.checkpoint_bundle_for(&heads)?))
            }
            Endpoint::Shallow => {
                let (heads, depth) = wire::decode_shallow(&body)?;
                ok(wire::encode_hashes(&r.shallow_changes(&heads, depth)?))
            }
            Endpoint::Push => {
                let (bundle, updates, policy, tags) = wire::decode_push(&body)?;
                let lock = self.lock_for(name);
                let _guard = lock.lock().unwrap();
                if r.is_unbound() {
                    // The first push binds the repository to the pusher's id (spec §16.1).
                    match claimed {
                        Some(id) => r.bind_id(id)?,
                        None => {
                            return Err(Failure(
                                StatusCode::BAD_REQUEST,
                                format!(
                                    "this repository is unbound; push with a client that sends {}",
                                    oxy_core::protocol::REPO_ID_HEADER
                                ),
                            ))
                        }
                    }
                }
                let protected = self.backend.default_bookmark(name);
                let (refused, updates): (Vec<_>, Vec<_>) = updates.into_iter().partition(|u| {
                    u.new_state.is_zero() && protected.as_deref() == Some(u.name.as_str())
                });
                let mut report = r.receive_push_report(bundle, updates, policy, tags)?;
                for u in refused {
                    report.results.push(oxy_core::protocol::RefResult {
                        name: u.name.clone(),
                        ok: false,
                        reason: format!(
                            "{:?} is this repository's default bookmark and cannot be deleted",
                            u.name
                        ),
                    });
                }
                self.backend.after_push(name, &report, headers);
                ok(wire::encode_push_response(&report.results))
            }
        })
    }
}

/// The backend of `oxy serve`: one repository per subdirectory of a root,
/// open reads, and pushes gated by a list of bearer tokens.
pub struct DirBackend {
    pub root: PathBuf,
    pub tokens: Vec<String>,
}

impl Backend for DirBackend {
    fn open(&self, name: &str) -> Result<Repo, Failure> {
        if !valid_repo_name(name) {
            return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
        }
        let dir = self.root.join(name);
        if !dir.join(".oxy").is_dir() {
            return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
        }
        Repo::open(&dir).map_err(|e| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
    }

    fn default_bookmark(&self, name: &str) -> Option<String> {
        self.open(name).ok().map(|r| r.config.bookmark.clone())
    }

    fn authorize(&self, _name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure> {
        if access == Access::Read || self.tokens.is_empty() {
            return Ok(());
        }
        match bearer_token(headers) {
            Some(t) if self.tokens.iter().any(|x| x == t) => Ok(()),
            _ => Err(Failure(
                StatusCode::UNAUTHORIZED,
                "a valid bearer token is required to push".into(),
            )),
        }
    }
}

fn hex_id(h: &str) -> Option<[u8; 16]> {
    let h = h.trim();
    if h.len() != 32 {
        return None;
    }
    let mut out = [0u8; 16];
    for (i, byte) in out.iter_mut().enumerate() {
        *byte = u8::from_str_radix(&h[2 * i..2 * i + 2], 16).ok()?;
    }
    Some(out)
}

/// The token of an `Authorization: Bearer` header, if any.
pub fn bearer_token(headers: &HeaderMap) -> Option<&str> {
    headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(str::trim)
}

async fn dispatch(
    State(server): State<Arc<Server>>,
    Path((repo, endpoint)): Path<(String, String)>,
    method: Method,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    match Endpoint::parse(&endpoint) {
        Some(e) => server.handle(repo, e, &method, &headers, body).await,
        None => Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response(),
    }
}

/// The router of `oxy serve`: `/{repo}/oxy/v1/{endpoint}` over a [`DirBackend`].
pub fn router(cfg: ServerConfig) -> Router {
    let server = Server::new(DirBackend {
        root: cfg.root,
        tokens: cfg.tokens,
    });
    Router::new()
        .route("/{repo}/oxy/v1/{endpoint}", any(dispatch))
        .with_state(server)
}

/// Serve on an already-bound listener until the process ends; HTTPS when
/// `cfg.tls` is set.
pub async fn serve(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
    listener.set_nonblocking(true)?;
    match cfg.tls.clone() {
        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 => {
            let listener = tokio::net::TcpListener::from_std(listener)?;
            axum::serve(listener, router(cfg)).await
        }
    }
}

/// Build a runtime and serve; for the CLI and for tests on a background thread.
pub fn run_blocking(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?;
    rt.block_on(serve(listener, cfg))
}

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

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	//! over TLS when a certificate and key are given, else plain HTTP for use
5	//! behind a TLS-terminating proxy or on a loopback interface (spec §15.1).
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	//! Hosts embed the protocol by implementing [`Backend`] (name resolution
11	//! and authorization) and routing to [`Server::handle`]; `router` is the
12	//! standalone server built on [`DirBackend`].
13
14	use axum::body::Bytes;
15	use axum::extract::{Path, State};
16	use axum::http::{header, HeaderMap, Method, StatusCode};
17	use axum::response::{IntoResponse, Response};
18	use axum::routing::any;
19	use axum::Router;
20	use oxy_core::protocol::{wire, CONTENT_TYPE};
21	use oxy_core::repo::Repo;
22	use std::collections::HashMap;
23	use std::path::PathBuf;
24	use std::sync::{Arc, Mutex};
25
26	#[derive(Clone, Debug)]
27	pub struct ServerConfig {
28	/// Directory containing one repository per subdirectory.
29	pub root: PathBuf,
30	/// Bearer tokens allowed to push; empty means no authentication.
31	pub tokens: Vec<String>,
32	/// Serve HTTPS with this certificate chain and private key (PEM files).
33	pub tls: Option<TlsFiles>,
34	}
35
36	#[derive(Clone, Debug)]
37	pub struct TlsFiles {
38	pub cert: PathBuf,
39	pub key: PathBuf,
40	}
41
42	/// Which kind of access a request needs (spec §15.1: reads may be open;
43	/// a push must be authenticated when any credential is configured).
44	#[derive(Clone, Copy, Debug, PartialEq, Eq)]
45	pub enum Access {
46	Read,
47	Push,
48	}
49
50	/// What a host provides to the protocol server: how a repository name maps
51	/// to a repository on disk and who may do what. `oxy serve` uses
52	/// [`DirBackend`]; a hub supplies its own with accounts and roles.
53	pub trait Backend: Send + Sync + 'static {
54	/// Open the repository called `name`, or fail with the status to
55	/// answer. An unknown name is `404`; a name the caller may not see
56	/// SHOULD also be `404`, so private repositories are not enumerable.
57	fn open(&self, name: &str) -> Result<Repo, Failure>;
58
59	/// Decide whether this request may perform `access` on `name`.
60	fn authorize(&self, name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure>;
61
62	/// Called after a push was processed, with its per-reference results
63	/// (accepted or refused), the states each accepted reference moved
64	/// between, and the changes stored, for logs and cache invalidation.
65	fn after_push(&self, _name: &str, _report: &oxy_core::repo::PushReport, _headers: &HeaderMap) {}
66
67	/// The bookmark of `name` that pushes may never delete (its default),
68	/// when the host keeps one apart from the repository's own config.
69	fn default_bookmark(&self, _name: &str) -> Option<String> {
70	None
71	}
72	}
73
74	/// A protocol server over any [`Backend`], serialising the pushes to each
75	/// repository. Hosts route requests to [`Server::handle`].
76	pub struct Server {
77	backend: Box<dyn Backend>,
78	locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
79	}
80
81	/// An error response: a status and a plain-text body.
82	#[derive(Debug)]
83	pub struct Failure(pub StatusCode, pub String);
84
85	impl IntoResponse for Failure {
86	fn into_response(self) -> Response {
87	(self.0, self.1).into_response()
88	}
89	}
90
91	impl From<oxy_core::repo::RepoError> for Failure {
92	fn from(e: oxy_core::repo::RepoError) -> Failure {
93	Failure(StatusCode::BAD_REQUEST, e.to_string())
94	}
95	}
96
97	/// The wire protocol's endpoints (spec §16.1), named as in the URL.
98	#[derive(Clone, Copy, Debug, PartialEq, Eq)]
99	pub enum Endpoint {
100	Capabilities,
101	Refs,
102	Negotiate,
103	Fetch,
104	Objects,
105	Checkpoint,
106	Shallow,
107	Push,
108	}
109
110	impl Endpoint {
111	pub fn parse(name: &str) -> Option<Endpoint> {
112	Some(match name {
113	"capabilities" => Endpoint::Capabilities,
114	"refs" => Endpoint::Refs,
115	"negotiate" => Endpoint::Negotiate,
116	"fetch" => Endpoint::Fetch,
117	"objects" => Endpoint::Objects,
118	"checkpoint" => Endpoint::Checkpoint,
119	"shallow" => Endpoint::Shallow,
120	"push" => Endpoint::Push,
121	_ => return None,
122	})
123	}
124
125	pub fn name(self) -> &'static str {
126	match self {
127	Endpoint::Capabilities => "capabilities",
128	Endpoint::Refs => "refs",
129	Endpoint::Negotiate => "negotiate",
130	Endpoint::Fetch => "fetch",
131	Endpoint::Objects => "objects",
132	Endpoint::Checkpoint => "checkpoint",
133	Endpoint::Shallow => "shallow",
134	Endpoint::Push => "push",
135	}
136	}
137
138	/// The HTTP method the endpoint answers to.
139	pub fn method(self) -> Method {
140	match self {
141	Endpoint::Capabilities \| Endpoint::Refs => Method::GET,
142	_ => Method::POST,
143	}
144	}
145
146	fn access(self) -> Access {
147	match self {
148	Endpoint::Push => Access::Push,
149	_ => Access::Read,
150	}
151	}
152	}
153
154	fn ok(body: Vec<u8>) -> Response {
155	([(header::CONTENT_TYPE, CONTENT_TYPE)], body).into_response()
156	}
157
158	pub fn valid_repo_name(name: &str) -> bool {
159	!name.is_empty()
160	&& !name.starts_with('.')
161	&& name
162	.chars()
163	.all(\|c\| c.is_ascii_alphanumeric() \|\| matches!(c, '.' \| '_' \| '-'))
164	}
165
166	impl Server {
167	pub fn new(backend: impl Backend) -> Arc<Server> {
168	Arc::new(Server {
169	backend: Box::new(backend),
170	locks: Mutex::new(HashMap::new()),
171	})
172	}
173
174	pub fn backend(&self) -> &dyn Backend {
175	self.backend.as_ref()
176	}
177
178	fn lock_for(&self, name: &str) -> Arc<Mutex<()>> {
179	self.locks
180	.lock()
181	.unwrap()
182	.entry(name.to_string())
183	.or_default()
184	.clone()
185	}
186
187	/// Answer one protocol request for repository `name`. `method` is
188	/// checked against the endpoint; the body is the raw CBOR request.
189	pub async fn handle(
190	self: &Arc<Self>,
191	name: String,
192	endpoint: Endpoint,
193	method: &Method,
194	headers: &HeaderMap,
195	body: Bytes,
196	) -> Response {
197	if *method != endpoint.method() {
198	return Failure(
199	StatusCode::METHOD_NOT_ALLOWED,
200	format!("{} is {}", endpoint.name(), endpoint.method()),
201	)
202	.into_response();
203	}
204	if let Err(f) = self.backend.authorize(&name, endpoint.access(), headers) {
205	return f.into_response();
206	}
207	let claimed = headers
208	.get(oxy_core::protocol::REPO_ID_HEADER)
209	.and_then(\|v\| v.to_str().ok())
210	.and_then(hex_id);
211	let server = Arc::clone(self);
212	let request_headers = headers.clone();
213	let result = tokio::task::spawn_blocking(move \|\| {
214	server.respond(&name, endpoint, body, claimed, &request_headers)
215	})
216	.await;
217	match result {
218	Ok(Ok(r)) => r,
219	Ok(Err(f)) => f.into_response(),
220	Err(e) => Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
221	}
222	}
223
224	fn respond(
225	&self,
226	name: &str,
227	endpoint: Endpoint,
228	body: Bytes,
229	claimed: Option<[u8; 16]>,
230	headers: &HeaderMap,
231	) -> Result<Response, Failure> {
232	let r = self.backend.open(name)?;
233	Ok(match endpoint {
234	Endpoint::Capabilities => ok(wire::encode_capabilities(r.id()?)),
235	Endpoint::Refs => ok(wire::encode_refs(r.id()?, &r.remote_refs()?)),
236	Endpoint::Negotiate => {
237	let (have, want) = wire::decode_negotiate(&body)?;
238	ok(wire::encode_hashes(&r.negotiate(&have, &want)?))
239	}
240	Endpoint::Fetch => {
241	let (changes, max_blob) = wire::decode_fetch(&body)?;
242	ok(wire::encode_bundle(
243	&r.bundle_for_filtered(&changes, max_blob)?,
244	))
245	}
246	Endpoint::Objects => {
247	let refs = wire::decode_hashes(&body)?;
248	let bundle = oxy_core::protocol::Bundle {
249	objects: r.objects_for(&refs)?,
250	..Default::default()
251	};
252	ok(wire::encode_bundle(&bundle))
253	}
254	Endpoint::Checkpoint => {
255	let heads = wire::decode_hashes(&body)?;
256	let lock = self.lock_for(name);
257	let _guard = lock.lock().unwrap();
258	ok(wire::encode_bundle(&r.checkpoint_bundle_for(&heads)?))
259	}
260	Endpoint::Shallow => {
261	let (heads, depth) = wire::decode_shallow(&body)?;
262	ok(wire::encode_hashes(&r.shallow_changes(&heads, depth)?))
263	}
264	Endpoint::Push => {
265	let (bundle, updates, policy, tags) = wire::decode_push(&body)?;
266	let lock = self.lock_for(name);
267	let _guard = lock.lock().unwrap();
268	if r.is_unbound() {
269	// The first push binds the repository to the pusher's id (spec §16.1).
270	match claimed {
271	Some(id) => r.bind_id(id)?,
272	None => {
273	return Err(Failure(
274	StatusCode::BAD_REQUEST,
275	format!(
276	"this repository is unbound; push with a client that sends {}",
277	oxy_core::protocol::REPO_ID_HEADER
278	),
279	))
280	}
281	}
282	}
283	let protected = self.backend.default_bookmark(name);
284	let (refused, updates): (Vec<_>, Vec<_>) = updates.into_iter().partition(\|u\| {
285	u.new_state.is_zero() && protected.as_deref() == Some(u.name.as_str())
286	});
287	let mut report = r.receive_push_report(bundle, updates, policy, tags)?;
288	for u in refused {
289	report.results.push(oxy_core::protocol::RefResult {
290	name: u.name.clone(),
291	ok: false,
292	reason: format!(
293	"{:?} is this repository's default bookmark and cannot be deleted",
294	u.name
295	),
296	});
297	}
298	self.backend.after_push(name, &report, headers);
299	ok(wire::encode_push_response(&report.results))
300	}
301	})
302	}
303	}
304
305	/// The backend of `oxy serve`: one repository per subdirectory of a root,
306	/// open reads, and pushes gated by a list of bearer tokens.
307	pub struct DirBackend {
308	pub root: PathBuf,
309	pub tokens: Vec<String>,
310	}
311
312	impl Backend for DirBackend {
313	fn open(&self, name: &str) -> Result<Repo, Failure> {
314	if !valid_repo_name(name) {
315	return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
316	}
317	let dir = self.root.join(name);
318	if !dir.join(".oxy").is_dir() {
319	return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
320	}
321	Repo::open(&dir).map_err(\|e\| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
322	}
323
324	fn default_bookmark(&self, name: &str) -> Option<String> {
325	self.open(name).ok().map(\|r\| r.config.bookmark.clone())
326	}
327
328	fn authorize(&self, _name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure> {
329	if access == Access::Read \|\| self.tokens.is_empty() {
330	return Ok(());
331	}
332	match bearer_token(headers) {
333	Some(t) if self.tokens.iter().any(\|x\| x == t) => Ok(()),
334	_ => Err(Failure(
335	StatusCode::UNAUTHORIZED,
336	"a valid bearer token is required to push".into(),
337	)),
338	}
339	}
340	}
341
342	fn hex_id(h: &str) -> Option<[u8; 16]> {
343	let h = h.trim();
344	if h.len() != 32 {
345	return None;
346	}
347	let mut out = [0u8; 16];
348	for (i, byte) in out.iter_mut().enumerate() {
349	byte = u8::from_str_radix(&h[2 i..2 * i + 2], 16).ok()?;
350	}
351	Some(out)
352	}
353
354	/// The token of an `Authorization: Bearer` header, if any.
355	pub fn bearer_token(headers: &HeaderMap) -> Option<&str> {
356	headers
357	.get(header::AUTHORIZATION)
358	.and_then(\|v\| v.to_str().ok())
359	.and_then(\|v\| v.strip_prefix("Bearer "))
360	.map(str::trim)
361	}
362
363	async fn dispatch(
364	State(server): State<Arc<Server>>,
365	Path((repo, endpoint)): Path<(String, String)>,
366	method: Method,
367	headers: HeaderMap,
368	body: Bytes,
369	) -> Response {
370	match Endpoint::parse(&endpoint) {
371	Some(e) => server.handle(repo, e, &method, &headers, body).await,
372	None => Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response(),
373	}
374	}
375
376	/// The router of `oxy serve`: `/{repo}/oxy/v1/{endpoint}` over a [`DirBackend`].
377	pub fn router(cfg: ServerConfig) -> Router {
378	let server = Server::new(DirBackend {
379	root: cfg.root,
380	tokens: cfg.tokens,
381	});
382	Router::new()
383	.route("/{repo}/oxy/v1/{endpoint}", any(dispatch))
384	.with_state(server)
385	}
386
387	/// Serve on an already-bound listener until the process ends; HTTPS when
388	/// `cfg.tls` is set.
389	pub async fn serve(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
390	listener.set_nonblocking(true)?;
391	match cfg.tls.clone() {
392	Some(tls) => {
393	let _ = rustls::crypto::ring::default_provider().install_default();
394	let rustls_cfg =
395	axum_server::tls_rustls::RustlsConfig::from_pem_file(&tls.cert, &tls.key)
396	.await
397	.map_err(\|e\| std::io::Error::other(format!("loading TLS files: {e}")))?;
398	axum_server::from_tcp_rustls(listener, rustls_cfg)?
399	.serve(router(cfg).into_make_service())
400	.await
401	}
402	None => {
403	let listener = tokio::net::TcpListener::from_std(listener)?;
404	axum::serve(listener, router(cfg)).await
405	}
406	}
407	}
408
409	/// Build a runtime and serve; for the CLI and for tests on a background thread.
410	pub fn run_blocking(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
411	let rt = tokio::runtime::Builder::new_multi_thread()
412	.enable_all()
413	.build()?;
414	rt.block_on(serve(listener, cfg))
415	}

Keyboard