OXYGENHUB
# de505df5cf8d oxygenhub / crates / hub / src / protocol.rs
8.2 KiB · Rust · last changed Milestone 0.2: accounts and access 1 month ago source blame history raw
d23464c0 Andy McCall ↶ 1
//! The wire protocol, embedded (hub spec §3.3, §5.4): `oxy-server`'s
19bb31ff Andy McCall ↶ 2
//! `Server` over a backend that resolves `owner/name`, authorises reads
3
//! and pushes by account role and token scope, and records who pushed.
d23464c0 Andy McCall ↶ 4
19bb31ff Andy McCall ↶ 5
use crate::auth::{self, Access, Role};
6
use crate::db::{self, accounts::Account};
d23464c0 Andy McCall ↶ 7
use crate::repos::{RepoName, Repos};
19bb31ff Andy McCall ↶ 8
use crate::Shared;
d23464c0 Andy McCall ↶ 9
use axum::body::Bytes;
10
use axum::extract::{Path, State};
11
use axum::http::{HeaderMap, Method, StatusCode};
12
use axum::response::{IntoResponse, Response};
62d6e969 Andy McCall ↶ 13
use oxy_core::repo::{PushReport, Repo};
19bb31ff Andy McCall ↶ 14
use oxy_server::{Backend, Endpoint, Failure};
d23464c0 Andy McCall ↶ 15
use std::sync::Arc;
16
17
pub struct HubBackend {
18
    repos: Arc<Repos>,
19
    pool: db::Pool,
19bb31ff Andy McCall ↶ 20
    /// Site tokens from `hub.toml`: full access, for bootstrapping.
21
    site_tokens: Vec<String>,
d23464c0 Andy McCall ↶ 22
    runtime: tokio::runtime::Handle,
23
}
24
19bb31ff Andy McCall ↶ 25
/// Who a request's credential resolved to.
26
enum Caller {
27
    Anonymous,
28
    /// A site token from the configuration.
29
    Site,
30
    Token(Box<Account>, Box<db::accounts::Token>),
31
}
32
d23464c0 Andy McCall ↶ 33
impl HubBackend {
19bb31ff Andy McCall ↶ 34
    pub fn new(repos: Arc<Repos>, pool: db::Pool, site_tokens: Vec<String>, runtime: tokio::runtime::Handle) -> HubBackend {
35
        HubBackend { repos, pool, site_tokens, runtime }
36
    }
37
38
    fn block<T>(&self, f: impl std::future::Future<Output = T>) -> T {
39
        tokio::task::block_in_place(|| self.runtime.block_on(f))
d23464c0 Andy McCall ↶ 40
    }
41
42
    fn row(&self, name: &RepoName) -> Result<db::repos::RepoRow, Failure> {
43
        let pool = self.pool.clone();
19bb31ff Andy McCall ↶ 44
        match self.block(db::repos::get(&pool, &name.owner, &name.name)) {
d23464c0 Andy McCall ↶ 45
            Ok(Some(r)) => Ok(r),
46
            Ok(None) => Err(not_found()),
47
            Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
48
        }
49
    }
50
19bb31ff Andy McCall ↶ 51
    fn caller(&self, headers: &HeaderMap) -> Result<Caller, Failure> {
52
        let Some(presented) = oxy_server::bearer_token(headers) else { return Ok(Caller::Anonymous) };
53
        if self.site_tokens.iter().any(|t| auth::mac_equal(t, presented)) {
54
            return Ok(Caller::Site);
d23464c0 Andy McCall ↶ 55
        }
19bb31ff Andy McCall ↶ 56
        let hash = auth::token_hash(presented);
57
        let pool = self.pool.clone();
58
        let token = self.block(db::accounts::token_by_hash(&pool, &hash)).map_err(internal)?;
59
        let Some(token) = token else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
60
        if token.expired(db::now_ms()) {
61
            return Err(Failure(StatusCode::UNAUTHORIZED, "this token has expired".into()));
62
        }
63
        let account = self.block(db::accounts::account_by_id(&pool, token.account_id)).map_err(internal)?;
64
        let Some(account) = account else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
65
        if account.banned {
66
            return Err(Failure(StatusCode::UNAUTHORIZED, "this account is banned".into()));
67
        }
68
        let _ = self.block(db::accounts::touch_token(&pool, token.id));
69
        Ok(Caller::Token(Box::new(account), Box::new(token)))
d23464c0 Andy McCall ↶ 70
    }
71
}
72
19bb31ff Andy McCall ↶ 73
fn internal(e: sqlx::Error) -> Failure {
74
    Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
75
}
76
d23464c0 Andy McCall ↶ 77
fn not_found() -> Failure {
78
    Failure(StatusCode::NOT_FOUND, "no such repository".into())
79
}
80
81
impl Backend for HubBackend {
82
    fn open(&self, name: &str) -> Result<Repo, Failure> {
83
        let name = RepoName::parse(name).ok_or_else(not_found)?;
84
        self.repos.open(&name).map_err(|e| match e {
85
            crate::repos::RepoError::NotFound => not_found(),
86
            other => Failure(StatusCode::INTERNAL_SERVER_ERROR, other.to_string()),
87
        })
88
    }
89
19bb31ff Andy McCall ↶ 90
    fn authorize(&self, name: &str, access: oxy_server::Access, headers: &HeaderMap) -> Result<(), Failure> {
d23464c0 Andy McCall ↶ 91
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
92
        let row = self.row(&parsed)?;
19bb31ff Andy McCall ↶ 93
        let caller = self.caller(headers)?;
94
        let (role, scope_ok) = match &caller {
95
            Caller::Site => (Some(Role::Admin), true),
96
            Caller::Anonymous => (self.block(auth::roles::role_for(&self.pool, None, &row)).map_err(internal)?, true),
97
            Caller::Token(account, token) => {
98
                let role = self.block(auth::roles::role_for(&self.pool, Some(account), &row)).map_err(internal)?;
99
                let needed = match access {
100
                    oxy_server::Access::Read => {
101
                        token.has_scope("repo:read") || token.has_scope("repo:write") || token.has_scope("repo:admin")
102
                    }
103
                    oxy_server::Access::Push => token.has_scope("repo:write") || token.has_scope("repo:admin"),
104
                };
105
                (role, needed && token.covers(&parsed.to_string()))
106
            }
107
        };
108
        // An unreadable repository does not exist for the caller (hub spec §4.4).
109
        if !auth::roles::allows(role, Access::Read) {
d23464c0 Andy McCall ↶ 110
            return Err(not_found());
111
        }
112
        match access {
19bb31ff Andy McCall ↶ 113
            oxy_server::Access::Read if scope_ok => Ok(()),
114
            oxy_server::Access::Read => Err(Failure(StatusCode::FORBIDDEN, "this token cannot read repositories".into())),
115
            oxy_server::Access::Push => {
116
                if matches!(caller, Caller::Anonymous) {
117
                    return Err(Failure(StatusCode::UNAUTHORIZED, "a personal access token is required to push".into()));
118
                }
119
                if !auth::roles::allows(role, Access::Write) {
120
                    return Err(Failure(StatusCode::FORBIDDEN, "you do not have write access to this repository".into()));
121
                }
122
                if !scope_ok {
123
                    return Err(Failure(StatusCode::FORBIDDEN, "this token lacks the repo:write scope for this repository".into()));
124
                }
125
                Ok(())
126
            }
d23464c0 Andy McCall ↶ 127
        }
128
    }
129
19bb31ff Andy McCall ↶ 130
    fn after_push(&self, name: &str, report: &PushReport, headers: &HeaderMap) {
d23464c0 Andy McCall ↶ 131
        let Some(parsed) = RepoName::parse(name) else { return };
132
        let Ok(row) = self.row(&parsed) else { return };
19bb31ff Andy McCall ↶ 133
        let pusher = match self.caller(headers) {
134
            Ok(Caller::Token(account, _)) => Some((account.id, account.username)),
135
            Ok(Caller::Site) => Some((0, "site token".to_string())),
136
            _ => None,
137
        };
d23464c0 Andy McCall ↶ 138
        let pool = self.pool.clone();
62d6e969 Andy McCall ↶ 139
        let results: Vec<db::repos::PushOutcome> = report
140
            .results
141
            .iter()
142
            .map(|r| {
143
                let updated = report.updated.iter().find(|u| u.name == r.name);
144
                db::repos::PushOutcome {
145
                    reference: r.name.clone(),
146
                    ok: r.ok,
147
                    reason: r.reason.clone(),
148
                    old_state: updated.map(|u| u.old_state.to_hex()).unwrap_or_default(),
149
                    new_state: updated.map(|u| u.new_state.to_hex()).unwrap_or_default(),
150
                    heads: updated.map(|u| u.heads.len() as i64).unwrap_or(0),
151
                    changes: if r.ok { report.imported.len() as i64 } else { 0 },
152
                }
153
            })
154
            .collect();
d23464c0 Andy McCall ↶ 155
        self.runtime.spawn(async move {
19bb31ff Andy McCall ↶ 156
            let who = pusher.as_ref().map(|(id, name)| (*id, name.as_str())).filter(|(id, _)| *id > 0);
157
            let label = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
158
            if let Err(e) =
159
                db::repos::record_push(&pool, row.id, who.or(if label.is_empty() { None } else { Some((0, label.as_str())) }), &results)
160
                    .await
161
            {
d23464c0 Andy McCall ↶ 162
                tracing::warn!("recording push to {}: {e}", parsed);
163
            }
164
        });
165
    }
166
}
167
168
/// `/{owner}/{name}/oxy/v1/{endpoint}`.
169
pub async fn dispatch(
170
    State(app): State<Shared>,
171
    Path((owner, name, endpoint)): Path<(String, String, String)>,
172
    method: Method,
173
    headers: HeaderMap,
174
    body: Bytes,
175
) -> Response {
176
    let Some(endpoint) = Endpoint::parse(&endpoint) else {
177
        return Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response();
178
    };
179
    let Some(repo) = RepoName::new(&owner, &name) else {
180
        return not_found().into_response();
181
    };
182
    let _guard = match endpoint {
183
        Endpoint::Push | Endpoint::Checkpoint => Guard::Write(app.repos.write_guard(&repo).await),
184
        _ => Guard::Read(app.repos.read_guard(&repo).await),
185
    };
186
    app.server.handle(repo.to_string(), endpoint, &method, &headers, body).await
187
}
188
189
enum Guard {
190
    Read(#[allow(dead_code)] tokio::sync::OwnedRwLockReadGuard<()>),
191
    Write(#[allow(dead_code)] tokio::sync::OwnedRwLockWriteGuard<()>),
192
}