crates/oxy-server/src/lib.rs at v0.7.0 · andymccall/oxygen

◆ v0.7.0 oxygen / crates / oxy-server / src / lib.rs

20.4 KiB · Rust · last changed oxy serve takes a push of more than 2 MB (closes #138) 10 days 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`].

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

    /// Vet a push by what it adds before anything of it is stored or any
    /// reference moves (spec §16.1): `bundle` is the push as decoded, its
    /// changes with their lines, `updates` the references it would move.
    /// A host refuses the whole push by returning a [`Failure`], whose
    /// text reaches the client as the push's error, as it is, several
    /// lines included; it should answer neither 401, 403 nor 404, which a
    /// client takes for a credential it should change. Called for every
    /// part of a push sent in parts (each part's changes are its own),
    /// outside the repository's push lock, so a slow vet holds up no
    /// other push. The default accepts everything at no cost.
    fn vet_push(
        &self,
        _name: &str,
        _repo: &Repo,
        _bundle: &oxy_core::protocol::Bundle,
        _updates: &[oxy_core::protocol::RefUpdate],
        _headers: &HeaderMap,
    ) -> Result<(), Failure> {
        Ok(())
    }

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

    /// Tailor a refs answer to the client asking, by its request headers
    /// (its `User-Agent` and the features it declares in `Oxy-Features`),
    /// after [`Backend::add_log`]. The default leaves out the tags a client
    /// that does not declare `certified-tags` would refuse
    /// ([`Repo::refs_for_client`], oxygen#131); a host that overrides it
    /// should apply that too ([`tailor_refs`]).
    fn adjust_refs(
        &self,
        _name: &str,
        repo: &Repo,
        refs: &mut oxy_core::protocol::RefsResponse,
        headers: &HeaderMap,
    ) {
        tailor_refs(repo, refs, headers);
    }

    /// Whether tag `tag` of `name` may be withdrawn (oxygenhub#301): called
    /// for each withdrawal a push carries, before the repository sees it.
    /// A host refuses one it depends on (a forge's release names the tag)
    /// with the reason, which is that withdrawal's result. The default
    /// allows every one; the policy is checked after, as for any push.
    fn vet_tag_withdrawal(&self, _name: &str, _tag: &str) -> Result<(), String> {
        Ok(())
    }

    /// 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);
                self.backend.adjust_refs(name, &r, &mut refs, headers);
                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();
                let bundle = r.checkpoint_bundle_for(&heads)?;
                let _t = oxy_core::timing::phase("serve: encode the checkpoint");
                ok(wire::encode_bundle(&bundle))
            }
            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)?;
                // The host vets what the push adds before anything is stored
                // or bound: a refused push leaves the repository as it was.
                self.backend
                    .vet_push(name, &r, &bundle, &updates, headers)?;
                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 kept_tags = Vec::new();
                let mut withdrawals_refused = Vec::new();
                for t in tags {
                    match t
                        .is_withdrawal()
                        .then(|| self.backend.vet_tag_withdrawal(name, &t.name))
                    {
                        Some(Err(reason)) => withdrawals_refused.push((t.name, reason)),
                        _ => kept_tags.push(t),
                    }
                }
                let mut report = r.receive_push_report(bundle, updates, policy, kept_tags)?;
                // The request named references, all refused here: it was no
                // push of objects alone, whatever the repository saw.
                if !refused.is_empty() || !withdrawals_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 (tag, reason) in withdrawals_refused {
                    report.results.push(oxy_core::protocol::RefResult {
                        name: format!("tags/{tag}"),
                        ok: false,
                        reason,
                    });
                }
                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 features a request declares in its `Oxy-Features` header, as the
/// header's text; `None` when it sent none (a client of the 0.5.0 era).
pub fn client_features(headers: &HeaderMap) -> Option<&str> {
    headers
        .get(oxy_core::protocol::FEATURES_HEADER)
        .and_then(|v| v.to_str().ok())
}

/// Tailor `refs` of `repo` to the client whose request carried `headers`:
/// the shared rule of [`Repo::refs_for_client`], which every host applies.
pub fn tailor_refs(repo: &Repo, refs: &mut oxy_core::protocol::RefsResponse, headers: &HeaderMap) {
    repo.refs_for_client(refs, client_features(headers));
}

/// 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 largest request `oxy serve` takes: a push part is about
/// `push.batch-bytes` (64 MiB), but one change can weigh more; OxygenHub's
/// default `push_body_limit` is the same.
pub const BODY_LIMIT: usize = if usize::BITS > 32 {
    4 << 30
} else {
    usize::MAX
};

/// 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,
    });
    // Without a limit of its own a request takes axum's 2 MB, and a larger
    // push was cut off with no word in the log ("Broken pipe" at the client).
    Router::new()
        .route("/{repo}/oxy/v1/{endpoint}", any(dispatch))
        .layer(axum::extract::DefaultBodyLimit::max(BODY_LIMIT))
        .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	fn after_push(&self, _name: &str, _report: &oxy_core::repo::PushReport, _headers: &HeaderMap) {}
68
69	/// Vet a push by what it adds before anything of it is stored or any
70	/// reference moves (spec §16.1): `bundle` is the push as decoded, its
71	/// changes with their lines, `updates` the references it would move.
72	/// A host refuses the whole push by returning a [`Failure`], whose
73	/// text reaches the client as the push's error, as it is, several
74	/// lines included; it should answer neither 401, 403 nor 404, which a
75	/// client takes for a credential it should change. Called for every
76	/// part of a push sent in parts (each part's changes are its own),
77	/// outside the repository's push lock, so a slow vet holds up no
78	/// other push. The default accepts everything at no cost.
79	fn vet_push(
80	&self,
81	_name: &str,
82	_repo: &Repo,
83	_bundle: &oxy_core::protocol::Bundle,
84	_updates: &[oxy_core::protocol::RefUpdate],
85	_headers: &HeaderMap,
86	) -> Result<(), Failure> {
87	Ok(())
88	}
89
90	/// Add the host's transparency log to a refs answer (spec §14.8): its
91	/// checkpoint, and each attestation's entry and inclusion proof. A host
92	/// without a log leaves it as it is.
93	fn add_log(&self, _name: &str, _refs: &mut oxy_core::protocol::RefsResponse) {}
94
95	/// Tailor a refs answer to the client asking, by its request headers
96	/// (its `User-Agent` and the features it declares in `Oxy-Features`),
97	/// after [`Backend::add_log`]. The default leaves out the tags a client
98	/// that does not declare `certified-tags` would refuse
99	/// ([`Repo::refs_for_client`], oxygen#131); a host that overrides it
100	/// should apply that too ([`tailor_refs`]).
101	fn adjust_refs(
102	&self,
103	_name: &str,
104	repo: &Repo,
105	refs: &mut oxy_core::protocol::RefsResponse,
106	headers: &HeaderMap,
107	) {
108	tailor_refs(repo, refs, headers);
109	}
110
111	/// Whether tag `tag` of `name` may be withdrawn (oxygenhub#301): called
112	/// for each withdrawal a push carries, before the repository sees it.
113	/// A host refuses one it depends on (a forge's release names the tag)
114	/// with the reason, which is that withdrawal's result. The default
115	/// allows every one; the policy is checked after, as for any push.
116	fn vet_tag_withdrawal(&self, _name: &str, _tag: &str) -> Result<(), String> {
117	Ok(())
118	}
119
120	/// The bookmark of `name` that pushes may never delete (its default),
121	/// when the host keeps one apart from the repository's own config.
122	fn default_bookmark(&self, _name: &str) -> Option<String> {
123	None
124	}
125	}
126
127	/// A protocol server over any [`Backend`], serialising the pushes to each
128	/// repository. Hosts route requests to [`Server::handle`].
129	pub struct Server {
130	backend: Box<dyn Backend>,
131	locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
132	}
133
134	/// An error response: a status and a plain-text body.
135	#[derive(Debug)]
136	pub struct Failure(pub StatusCode, pub String);
137
138	impl IntoResponse for Failure {
139	fn into_response(self) -> Response {
140	(self.0, self.1).into_response()
141	}
142	}
143
144	impl From<oxy_core::repo::RepoError> for Failure {
145	fn from(e: oxy_core::repo::RepoError) -> Failure {
146	use oxy_core::repo::RepoError;
147	// The server's own storage failing is not the client's fault.
148	let status = match &e {
149	RepoError::Io(_) \| RepoError::Store(_) => StatusCode::INTERNAL_SERVER_ERROR,
150	_ => StatusCode::BAD_REQUEST,
151	};
152	Failure(status, e.to_string())
153	}
154	}
155
156	/// The wire protocol's endpoints (spec §16.1), named as in the URL.
157	#[derive(Clone, Copy, Debug, PartialEq, Eq)]
158	pub enum Endpoint {
159	Capabilities,
160	Refs,
161	Negotiate,
162	Fetch,
163	Objects,
164	Checkpoint,
165	Shallow,
166	Push,
167	}
168
169	impl Endpoint {
170	pub fn parse(name: &str) -> Option<Endpoint> {
171	Some(match name {
172	"capabilities" => Endpoint::Capabilities,
173	"refs" => Endpoint::Refs,
174	"negotiate" => Endpoint::Negotiate,
175	"fetch" => Endpoint::Fetch,
176	"objects" => Endpoint::Objects,
177	"checkpoint" => Endpoint::Checkpoint,
178	"shallow" => Endpoint::Shallow,
179	"push" => Endpoint::Push,
180	_ => return None,
181	})
182	}
183
184	pub fn name(self) -> &'static str {
185	match self {
186	Endpoint::Capabilities => "capabilities",
187	Endpoint::Refs => "refs",
188	Endpoint::Negotiate => "negotiate",
189	Endpoint::Fetch => "fetch",
190	Endpoint::Objects => "objects",
191	Endpoint::Checkpoint => "checkpoint",
192	Endpoint::Shallow => "shallow",
193	Endpoint::Push => "push",
194	}
195	}
196
197	/// The HTTP method the endpoint answers to.
198	pub fn method(self) -> Method {
199	match self {
200	Endpoint::Capabilities \| Endpoint::Refs => Method::GET,
201	_ => Method::POST,
202	}
203	}
204
205	fn access(self) -> Access {
206	match self {
207	Endpoint::Push => Access::Push,
208	_ => Access::Read,
209	}
210	}
211	}
212
213	fn ok(body: Vec<u8>) -> Response {
214	([(header::CONTENT_TYPE, CONTENT_TYPE)], body).into_response()
215	}
216
217	pub fn valid_repo_name(name: &str) -> bool {
218	!name.is_empty()
219	&& !name.starts_with('.')
220	&& name
221	.chars()
222	.all(\|c\| c.is_ascii_alphanumeric() \|\| matches!(c, '.' \| '_' \| '-'))
223	}
224
225	impl Server {
226	pub fn new(backend: impl Backend) -> Arc<Server> {
227	Arc::new(Server {
228	backend: Box::new(backend),
229	locks: Mutex::new(HashMap::new()),
230	})
231	}
232
233	pub fn backend(&self) -> &dyn Backend {
234	self.backend.as_ref()
235	}
236
237	fn lock_for(&self, name: &str) -> Arc<Mutex<()>> {
238	self.locks
239	.lock()
240	.unwrap()
241	.entry(name.to_string())
242	.or_default()
243	.clone()
244	}
245
246	/// Answer one protocol request for repository `name`. `method` is
247	/// checked against the endpoint; the body is the raw CBOR request.
248	pub async fn handle(
249	self: &Arc<Self>,
250	name: String,
251	endpoint: Endpoint,
252	method: &Method,
253	headers: &HeaderMap,
254	body: Bytes,
255	) -> Response {
256	if *method != endpoint.method() {
257	return Failure(
258	StatusCode::METHOD_NOT_ALLOWED,
259	format!("{} is {}", endpoint.name(), endpoint.method()),
260	)
261	.into_response();
262	}
263	if let Err(f) = self.backend.authorize(&name, endpoint.access(), headers) {
264	return f.into_response();
265	}
266	let claimed = headers
267	.get(oxy_core::protocol::REPO_ID_HEADER)
268	.and_then(\|v\| v.to_str().ok())
269	.and_then(hex_id);
270	let server = Arc::clone(self);
271	let request_headers = headers.clone();
272	let repo = name.clone();
273	let result = tokio::task::spawn_blocking(move \|\| {
274	server.respond(&repo, endpoint, body, claimed, &request_headers)
275	})
276	.await;
277	let failure = match result {
278	Ok(Ok(r)) => return r,
279	Ok(Err(f)) => f,
280	Err(e) => Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()),
281	};
282	// Name the operation and repository, in the log and for the client.
283	let f = Failure(
284	failure.0,
285	format!("{} of {name}: {}", endpoint.name(), failure.1),
286	);
287	eprintln!("oxy-server: {}: {}", f.0.as_u16(), f.1);
288	f.into_response()
289	}
290
291	fn respond(
292	&self,
293	name: &str,
294	endpoint: Endpoint,
295	body: Bytes,
296	claimed: Option<[u8; 16]>,
297	headers: &HeaderMap,
298	) -> Result<Response, Failure> {
299	let r = self.backend.open(name)?;
300	Ok(match endpoint {
301	Endpoint::Capabilities => ok(wire::encode_capabilities(r.id()?)),
302	Endpoint::Refs => {
303	let mut refs = r.remote_refs()?;
304	if let Some(d) = self.backend.default_bookmark(name) {
305	refs.default_bookmark = Some(d);
306	}
307	self.backend.add_log(name, &mut refs);
308	self.backend.adjust_refs(name, &r, &mut refs, headers);
309	ok(wire::encode_refs(r.id()?, &refs))
310	}
311	Endpoint::Negotiate => {
312	let (have, want) = wire::decode_negotiate(&body)?;
313	ok(wire::encode_hashes(&r.negotiate(&have, &want)?))
314	}
315	Endpoint::Fetch => {
316	let (changes, max_blob) = wire::decode_fetch(&body)?;
317	ok(wire::encode_bundle(
318	&r.bundle_for_filtered(&changes, max_blob)?,
319	))
320	}
321	Endpoint::Objects => {
322	let refs = wire::decode_hashes(&body)?;
323	let bundle = oxy_core::protocol::Bundle {
324	objects: r.objects_for(&refs)?,
325	..Default::default()
326	};
327	ok(wire::encode_bundle(&bundle))
328	}
329	Endpoint::Checkpoint => {
330	let heads = wire::decode_hashes(&body)?;
331	let lock = self.lock_for(name);
332	let _guard = lock.lock().unwrap();
333	let bundle = r.checkpoint_bundle_for(&heads)?;
334	let _t = oxy_core::timing::phase("serve: encode the checkpoint");
335	ok(wire::encode_bundle(&bundle))
336	}
337	Endpoint::Shallow => {
338	let (heads, depth) = wire::decode_shallow(&body)?;
339	ok(wire::encode_hashes(&r.shallow_changes(&heads, depth)?))
340	}
341	Endpoint::Push => {
342	let (bundle, updates, policy, tags) = wire::decode_push(&body)?;
343	// The host vets what the push adds before anything is stored
344	// or bound: a refused push leaves the repository as it was.
345	self.backend
346	.vet_push(name, &r, &bundle, &updates, headers)?;
347	let lock = self.lock_for(name);
348	let _guard = lock.lock().unwrap();
349	if r.is_unbound() {
350	// The first push binds the repository to the pusher's id (spec §16.1).
351	match claimed {
352	Some(id) => r.bind_id(id)?,
353	None => {
354	return Err(Failure(
355	StatusCode::BAD_REQUEST,
356	format!(
357	"this repository is unbound; push with a client that sends {}",
358	oxy_core::protocol::REPO_ID_HEADER
359	),
360	))
361	}
362	}
363	}
364	let protected = self.backend.default_bookmark(name);
365	let (refused, updates): (Vec<_>, Vec<_>) = updates.into_iter().partition(\|u\| {
366	u.new_state.is_zero() && protected.as_deref() == Some(u.name.as_str())
367	});
368	let mut kept_tags = Vec::new();
369	let mut withdrawals_refused = Vec::new();
370	for t in tags {
371	match t
372	.is_withdrawal()
373	.then(\|\| self.backend.vet_tag_withdrawal(name, &t.name))
374	{
375	Some(Err(reason)) => withdrawals_refused.push((t.name, reason)),
376	_ => kept_tags.push(t),
377	}
378	}
379	let mut report = r.receive_push_report(bundle, updates, policy, kept_tags)?;
380	// The request named references, all refused here: it was no
381	// push of objects alone, whatever the repository saw.
382	if !refused.is_empty() \|\| !withdrawals_refused.is_empty() {
383	report
384	.results
385	.retain(\|r\| r.name != oxy_core::repo::OBJECTS_RESULT);
386	}
387	for u in refused {
388	report.results.push(oxy_core::protocol::RefResult {
389	name: u.name.clone(),
390	ok: false,
391	reason: format!(
392	"{:?} is this repository's default bookmark and cannot be deleted",
393	u.name
394	),
395	});
396	}
397	for (tag, reason) in withdrawals_refused {
398	report.results.push(oxy_core::protocol::RefResult {
399	name: format!("tags/{tag}"),
400	ok: false,
401	reason,
402	});
403	}
404	for r in report.results.iter().filter(\|r\| !r.ok) {
405	eprintln!(
406	"oxy-server: push to {name}: {} refused: {}",
407	r.name, r.reason
408	);
409	}
410	self.backend.after_push(name, &report, headers);
411	ok(wire::encode_push_response(&report.results))
412	}
413	})
414	}
415	}
416
417	/// The backend of `oxy serve`: one repository per subdirectory of a root,
418	/// open reads, and pushes gated by a list of bearer tokens.
419	pub struct DirBackend {
420	pub root: PathBuf,
421	pub tokens: Vec<String>,
422	}
423
424	impl Backend for DirBackend {
425	fn open(&self, name: &str) -> Result<Repo, Failure> {
426	if !valid_repo_name(name) {
427	return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
428	}
429	let dir = self.root.join(name);
430	if !dir.join(".oxy").is_dir() {
431	return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
432	}
433	Repo::open(&dir).map_err(\|e\| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
434	}
435
436	fn default_bookmark(&self, name: &str) -> Option<String> {
437	self.open(name).ok().map(\|r\| r.config.bookmark.clone())
438	}
439
440	fn authorize(&self, _name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure> {
441	if access == Access::Read \|\| self.tokens.is_empty() {
442	return Ok(());
443	}
444	match bearer_token(headers) {
445	Some(t) if self.tokens.iter().any(\|x\| x == t) => Ok(()),
446	_ => Err(Failure(
447	StatusCode::UNAUTHORIZED,
448	"a valid bearer token is required to push".into(),
449	)),
450	}
451	}
452	}
453
454	fn hex_id(h: &str) -> Option<[u8; 16]> {
455	let h = h.trim();
456	if h.len() != 32 {
457	return None;
458	}
459	let mut out = [0u8; 16];
460	for (i, byte) in out.iter_mut().enumerate() {
461	byte = u8::from_str_radix(&h[2 i..2 * i + 2], 16).ok()?;
462	}
463	Some(out)
464	}
465
466	/// The features a request declares in its `Oxy-Features` header, as the
467	/// header's text; `None` when it sent none (a client of the 0.5.0 era).
468	pub fn client_features(headers: &HeaderMap) -> Option<&str> {
469	headers
470	.get(oxy_core::protocol::FEATURES_HEADER)
471	.and_then(\|v\| v.to_str().ok())
472	}
473
474	/// Tailor `refs` of `repo` to the client whose request carried `headers`:
475	/// the shared rule of [`Repo::refs_for_client`], which every host applies.
476	pub fn tailor_refs(repo: &Repo, refs: &mut oxy_core::protocol::RefsResponse, headers: &HeaderMap) {
477	repo.refs_for_client(refs, client_features(headers));
478	}
479
480	/// The token of an `Authorization: Bearer` header, if any.
481	pub fn bearer_token(headers: &HeaderMap) -> Option<&str> {
482	headers
483	.get(header::AUTHORIZATION)
484	.and_then(\|v\| v.to_str().ok())
485	.and_then(\|v\| v.strip_prefix("Bearer "))
486	.map(str::trim)
487	}
488
489	async fn dispatch(
490	State(server): State<Arc<Server>>,
491	Path((repo, endpoint)): Path<(String, String)>,
492	method: Method,
493	headers: HeaderMap,
494	body: Bytes,
495	) -> Response {
496	match Endpoint::parse(&endpoint) {
497	Some(e) => server.handle(repo, e, &method, &headers, body).await,
498	None => Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response(),
499	}
500	}
501
502	/// The largest request `oxy serve` takes: a push part is about
503	/// `push.batch-bytes` (64 MiB), but one change can weigh more; OxygenHub's
504	/// default `push_body_limit` is the same.
505	pub const BODY_LIMIT: usize = if usize::BITS > 32 {
506	4 << 30
507	} else {
508	usize::MAX
509	};
510
511	/// The router of `oxy serve`: `/{repo}/oxy/v1/{endpoint}` over a [`DirBackend`].
512	pub fn router(cfg: ServerConfig) -> Router {
513	let server = Server::new(DirBackend {
514	root: cfg.root,
515	tokens: cfg.tokens,
516	});
517	// Without a limit of its own a request takes axum's 2 MB, and a larger
518	// push was cut off with no word in the log ("Broken pipe" at the client).
519	Router::new()
520	.route("/{repo}/oxy/v1/{endpoint}", any(dispatch))
521	.layer(axum::extract::DefaultBodyLimit::max(BODY_LIMIT))
522	.with_state(server)
523	}
524
525	/// Serve on an already-bound listener until the process ends; HTTPS when
526	/// `cfg.tls` is set.
527	pub async fn serve(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
528	listener.set_nonblocking(true)?;
529	match cfg.tls.clone() {
530	Some(tls) => {
531	let _ = rustls::crypto::ring::default_provider().install_default();
532	let rustls_cfg =
533	axum_server::tls_rustls::RustlsConfig::from_pem_file(&tls.cert, &tls.key)
534	.await
535	.map_err(\|e\| std::io::Error::other(format!("loading TLS files: {e}")))?;
536	axum_server::from_tcp_rustls(listener, rustls_cfg)?
537	.serve(router(cfg).into_make_service())
538	.await
539	}
540	None => {
541	let listener = tokio::net::TcpListener::from_std(listener)?;
542	axum::serve(listener, router(cfg)).await
543	}
544	}
545	}
546
547	/// Build a runtime and serve; for the CLI and for tests on a background thread.
548	pub fn run_blocking(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
549	let rt = tokio::runtime::Builder::new_multi_thread()
550	.enable_all()
551	.build()?;
552	rt.block_on(serve(listener, cfg))
553	}

Keyboard