lib.rs 85.7%

246 of 287 lines run, 38 of 83 functions · run #356 on main · run missed

//! 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`].

pub mod witness;

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) {}

    /// Add the host's transparency log to a refs answer (spec §14.8): its
    /// checkpoint, and each attestation's entry and inclusion proof. A host
    /// without a log leaves it as it is.
    fn add_log(&self, _name: &str, _refs: &mut oxy_core::protocol::RefsResponse) {}

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

Coverage / crates/oxy-server/src/lib.rs 85.7%

Keyboard