//! The wire protocol, embedded (hub spec §3.3, §5.4): `oxy-server`'s //! `Server` over a backend that resolves `owner/name`, applies visibility //! and push tokens, and records pushes. use crate::repos::{RepoName, Repos}; use crate::{db, Shared}; use axum::body::Bytes; use axum::extract::{Path, State}; use axum::http::{HeaderMap, Method, StatusCode}; use axum::response::{IntoResponse, Response}; use hub_api::Visibility; use oxy_core::protocol::RefResult; use oxy_core::repo::Repo; use oxy_server::{Access, Backend, Endpoint, Failure}; use std::sync::Arc; pub struct HubBackend { repos: Arc, pool: db::Pool, push_tokens: Vec, runtime: tokio::runtime::Handle, } impl HubBackend { pub fn new(repos: Arc, pool: db::Pool, push_tokens: Vec, runtime: tokio::runtime::Handle) -> HubBackend { HubBackend { repos, pool, push_tokens, runtime } } fn row(&self, name: &RepoName) -> Result { let pool = self.pool.clone(); let (owner, repo) = (name.owner.clone(), name.name.clone()); let row = tokio::task::block_in_place(|| self.runtime.block_on(db::repos::get(&pool, &owner, &repo))); match row { Ok(Some(r)) => Ok(r), Ok(None) => Err(not_found()), Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())), } } fn has_push_token(&self, headers: &HeaderMap) -> bool { match oxy_server::bearer_token(headers) { Some(t) => self.push_tokens.iter().any(|x| x == t), None => false, } } } fn not_found() -> Failure { Failure(StatusCode::NOT_FOUND, "no such repository".into()) } impl Backend for HubBackend { fn open(&self, name: &str) -> Result { let name = RepoName::parse(name).ok_or_else(not_found)?; self.repos.open(&name).map_err(|e| match e { crate::repos::RepoError::NotFound => not_found(), other => Failure(StatusCode::INTERNAL_SERVER_ERROR, other.to_string()), }) } fn authorize(&self, name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure> { let parsed = RepoName::parse(name).ok_or_else(not_found)?; let row = self.row(&parsed)?; let token = self.has_push_token(headers); // Until accounts exist (hub 0.2) a push token is the only credential: // it reads everything and may push everywhere. if row.visibility() != Visibility::Public && !token { return Err(not_found()); } match access { Access::Read => Ok(()), Access::Push if self.push_tokens.is_empty() || token => Ok(()), Access::Push => Err(Failure(StatusCode::UNAUTHORIZED, "a valid bearer token is required to push".into())), } } fn after_push(&self, name: &str, results: &[RefResult]) { let Some(parsed) = RepoName::parse(name) else { return }; let Ok(row) = self.row(&parsed) else { return }; let pool = self.pool.clone(); let results: Vec<(String, bool, String)> = results.iter().map(|r| (r.name.clone(), r.ok, r.reason.clone())).collect(); self.runtime.spawn(async move { if let Err(e) = db::repos::record_push(&pool, row.id, &results).await { tracing::warn!("recording push to {}: {e}", parsed); } }); } } /// `/{owner}/{name}/oxy/v1/{endpoint}`. pub async fn dispatch( State(app): State, Path((owner, name, endpoint)): Path<(String, String, String)>, method: Method, headers: HeaderMap, body: Bytes, ) -> Response { let Some(endpoint) = Endpoint::parse(&endpoint) else { return Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response(); }; let Some(repo) = RepoName::new(&owner, &name) else { return not_found().into_response(); }; // Pushes and checkpoints write; everything else reads. let _guard = match endpoint { Endpoint::Push | Endpoint::Checkpoint => Guard::Write(app.repos.write_guard(&repo).await), _ => Guard::Read(app.repos.read_guard(&repo).await), }; app.server.handle(repo.to_string(), endpoint, &method, &headers, body).await } enum Guard { Read(#[allow(dead_code)] tokio::sync::OwnedRwLockReadGuard<()>), Write(#[allow(dead_code)] tokio::sync::OwnedRwLockWriteGuard<()>), }