OXYGENHUB
# baebdd18aaa2 oxygenhub / crates / hub / src / protocol.rs
4.3 KiB · Rust · last changed Milestone 0.1: browse 1 month ago source blame history raw
1 
//! The wire protocol, embedded (hub spec §3.3, §5.4): `oxy-server`'s
2 
//! `Server` over a backend that resolves `owner/name`, applies visibility
3 
//! and push tokens, and records pushes.
4
5 
use crate::repos::{RepoName, Repos};
6 
use crate::{db, Shared};
7 
use axum::body::Bytes;
8 
use axum::extract::{Path, State};
9 
use axum::http::{HeaderMap, Method, StatusCode};
10 
use axum::response::{IntoResponse, Response};
11 
use hub_api::Visibility;
12 
use oxy_core::protocol::RefResult;
13 
use oxy_core::repo::Repo;
14 
use oxy_server::{Access, Backend, Endpoint, Failure};
15 
use std::sync::Arc;
16
17 
pub struct HubBackend {
18 
    repos: Arc<Repos>,
19 
    pool: db::Pool,
20 
    push_tokens: Vec<String>,
21 
    runtime: tokio::runtime::Handle,
22 
}
23
24 
impl HubBackend {
25 
    pub fn new(repos: Arc<Repos>, pool: db::Pool, push_tokens: Vec<String>, runtime: tokio::runtime::Handle) -> HubBackend {
26 
        HubBackend { repos, pool, push_tokens, runtime }
27 
    }
28
29 
    fn row(&self, name: &RepoName) -> Result<db::repos::RepoRow, Failure> {
30 
        let pool = self.pool.clone();
31 
        let (owner, repo) = (name.owner.clone(), name.name.clone());
32 
        let row = tokio::task::block_in_place(|| self.runtime.block_on(db::repos::get(&pool, &owner, &repo)));
33 
        match row {
34 
            Ok(Some(r)) => Ok(r),
35 
            Ok(None) => Err(not_found()),
36 
            Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
37 
        }
38 
    }
39
40 
    fn has_push_token(&self, headers: &HeaderMap) -> bool {
41 
        match oxy_server::bearer_token(headers) {
42 
            Some(t) => self.push_tokens.iter().any(|x| x == t),
43 
            None => false,
44 
        }
45 
    }
46 
}
47
48 
fn not_found() -> Failure {
49 
    Failure(StatusCode::NOT_FOUND, "no such repository".into())
50 
}
51
52 
impl Backend for HubBackend {
53 
    fn open(&self, name: &str) -> Result<Repo, Failure> {
54 
        let name = RepoName::parse(name).ok_or_else(not_found)?;
55 
        self.repos.open(&name).map_err(|e| match e {
56 
            crate::repos::RepoError::NotFound => not_found(),
57 
            other => Failure(StatusCode::INTERNAL_SERVER_ERROR, other.to_string()),
58 
        })
59 
    }
60
61 
    fn authorize(&self, name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure> {
62 
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
63 
        let row = self.row(&parsed)?;
64 
        let token = self.has_push_token(headers);
65 
        // Until accounts exist (hub 0.2) a push token is the only credential:
66 
        // it reads everything and may push everywhere.
67 
        if row.visibility() != Visibility::Public && !token {
68 
            return Err(not_found());
69 
        }
70 
        match access {
71 
            Access::Read => Ok(()),
72 
            Access::Push if self.push_tokens.is_empty() || token => Ok(()),
73 
            Access::Push => Err(Failure(StatusCode::UNAUTHORIZED, "a valid bearer token is required to push".into())),
74 
        }
75 
    }
76
77 
    fn after_push(&self, name: &str, results: &[RefResult]) {
78 
        let Some(parsed) = RepoName::parse(name) else { return };
79 
        let Ok(row) = self.row(&parsed) else { return };
80 
        let pool = self.pool.clone();
81 
        let results: Vec<(String, bool, String)> = results.iter().map(|r| (r.name.clone(), r.ok, r.reason.clone())).collect();
82 
        self.runtime.spawn(async move {
83 
            if let Err(e) = db::repos::record_push(&pool, row.id, &results).await {
84 
                tracing::warn!("recording push to {}: {e}", parsed);
85 
            }
86 
        });
87 
    }
88 
}
89
90 
/// `/{owner}/{name}/oxy/v1/{endpoint}`.
91 
pub async fn dispatch(
92 
    State(app): State<Shared>,
93 
    Path((owner, name, endpoint)): Path<(String, String, String)>,
94 
    method: Method,
95 
    headers: HeaderMap,
96 
    body: Bytes,
97 
) -> Response {
98 
    let Some(endpoint) = Endpoint::parse(&endpoint) else {
99 
        return Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response();
100 
    };
101 
    let Some(repo) = RepoName::new(&owner, &name) else {
102 
        return not_found().into_response();
103 
    };
104 
    // Pushes and checkpoints write; everything else reads.
105 
    let _guard = match endpoint {
106 
        Endpoint::Push | Endpoint::Checkpoint => Guard::Write(app.repos.write_guard(&repo).await),
107 
        _ => Guard::Read(app.repos.read_guard(&repo).await),
108 
    };
109 
    app.server.handle(repo.to_string(), endpoint, &method, &headers, body).await
110 
}
111
112 
enum Guard {
113 
    Read(#[allow(dead_code)] tokio::sync::OwnedRwLockReadGuard<()>),
114 
    Write(#[allow(dead_code)] tokio::sync::OwnedRwLockWriteGuard<()>),
115 
}