OXYGENHUB
# a549ddd5941f oxygenhub / crates / hub / src / protocol.rs
30.3 KiB · Rust · last changed Hub: old clients are not sent tags only a certificate authorizes (refs oxygen#131) 15 days 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`, authorises reads
3 
//! and pushes by account role and token scope, and records who pushed.
4
5 
use crate::auth::{self, Access, Role};
6 
use crate::db::{self, accounts::Account};
7 
use crate::repos::{RepoName, Repos};
8 
use crate::Shared;
9 
use axum::body::Bytes;
10 
use axum::extract::{Path, State};
11 
use axum::http::{HeaderMap, Method, StatusCode};
12 
use axum::response::{IntoResponse, Response};
13 
use oxy_core::repo::{PushReport, Repo};
14 
use oxy_server::{Backend, Endpoint, Failure};
15 
use std::sync::Arc;
16
17 
pub struct HubBackend {
18 
    repos: Arc<Repos>,
19 
    /// Wikis, reached as `OWNER/NAME.wiki` (hub spec §6.11).
20 
    wikis: Arc<Repos>,
21 
    pool: db::Pool,
22 
    /// Site tokens from `hub.toml`: full access, for bootstrapping.
23 
    site_tokens: Vec<String>,
24 
    runtime: tokio::runtime::Handle,
25 
    /// The application, set once it exists (the backend is built first).
26 
    app_hint: Arc<std::sync::Mutex<Option<std::sync::Weak<crate::App>>>>,
27 
    /// The hub's key, which countersigns when it accepted an update a key
28 
    /// it certified signed (oxygenhub#241 step 4c, Oxygen spec §14.7).
29 
    acceptor: Option<Arc<(oxy_core::identity::Identity, oxy_core::identity::SecretKey)>>,
30 
    /// Push protection's scan budget (#268, `[limits]`).
31 
    scan_budget: crate::push_protection::Budget,
32 
    /// Repositories whose push protection scan stopped early, by id: the
33 
    /// note goes on the push log entry of the push that moves references
34 
    /// (the parts of a long push before it move none).
35 
    scan_notes: Arc<std::sync::Mutex<std::collections::HashMap<i64, String>>>,
36 
}
37
38 
/// Who a request's credential resolved to.
39 
enum Caller {
40 
    Anonymous,
41 
    /// A site token from the configuration.
42 
    Site,
43 
    Token(Box<Account>, Box<db::accounts::Token>),
44 
    /// A running OxygenFlow job: reads its own repository and the subrepos
45 
    /// its state pins, nothing else (hub spec §13.2).
46 
    Job {
47 
        repo_id: i64,
48 
        also: Vec<i64>,
49 
    },
50 
}
51
52 
impl HubBackend {
53 
    pub fn new(
54 
        repos: Arc<Repos>,
55 
        wikis: Arc<Repos>,
56 
        pool: db::Pool,
57 
        site_tokens: Vec<String>,
58 
        runtime: tokio::runtime::Handle,
59 
    ) -> HubBackend {
60 
        HubBackend {
61 
            repos,
62 
            wikis,
63 
            pool,
64 
            site_tokens,
65 
            runtime,
66 
            app_hint: Arc::new(std::sync::Mutex::new(None)),
67 
            acceptor: None,
68 
            scan_budget: crate::push_protection::Budget::default(),
69 
            scan_notes: Arc::default(),
70 
        }
71 
    }
72
73 
    /// Scan pushes for secrets within `budget` (#268).
74 
    pub fn with_scan_budget(mut self, budget: crate::push_protection::Budget) -> Self {
75 
        self.scan_budget = budget;
76 
        self
77 
    }
78
79 
    /// Who pushes, for the push log: the account (id 0 for a site token)
80 
    /// and, for an agent's token, the agent's name.
81 
    fn pusher(&self, headers: &HeaderMap) -> (Option<(i64, String)>, String) {
82 
        match self.caller(headers) {
83 
            Ok(Caller::Token(account, token)) => {
84 
                (Some((account.id, account.username)), if token.agent { token.name.clone() } else { String::new() })
85 
            }
86 
            Ok(Caller::Site) => (Some((0, "site token".to_string())), String::new()),
87 
            _ => (None, String::new()),
88 
        }
89 
    }
90
91 
    /// Countersign accepted updates by keys this hub certified with `identity`.
92 
    pub fn with_acceptor(mut self, identity: oxy_core::identity::Identity, secret: oxy_core::identity::SecretKey) -> Self {
93 
        self.acceptor = Some(Arc::new((identity, secret)));
94 
        self
95 
    }
96
97 
    /// The slot the application registers itself in after construction.
98 
    pub fn app_slot(&self) -> Arc<std::sync::Mutex<Option<std::sync::Weak<crate::App>>>> {
99 
        self.app_hint.clone()
100 
    }
101
102 
    fn block<T>(&self, f: impl std::future::Future<Output = T>) -> T {
103 
        tokio::task::block_in_place(|| self.runtime.block_on(f))
104 
    }
105
106 
    fn row(&self, name: &RepoName) -> Result<db::repos::RepoRow, Failure> {
107 
        let pool = self.pool.clone();
108 
        match self.block(db::repos::get(&pool, &name.owner, &name.name)) {
109 
            Ok(Some(r)) => Ok(r),
110 
            // A transferred repository still answers to its old name (#108).
111 
            Ok(None) => match self.block(db::repos::moved_to(&pool, &name.owner, &name.name)) {
112 
                Ok(Some(r)) => Ok(r),
113 
                Ok(None) => Err(not_found()),
114 
                Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
115 
            },
116 
            Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
117 
        }
118 
    }
119
120 
    fn caller(&self, headers: &HeaderMap) -> Result<Caller, Failure> {
121 
        let Some(presented) = oxy_server::bearer_token(headers) else { return Ok(Caller::Anonymous) };
122 
        if self.site_tokens.iter().any(|t| auth::mac_equal(t, presented)) {
123 
            return Ok(Caller::Site);
124 
        }
125 
        let hash = auth::token_hash(presented);
126 
        let pool = self.pool.clone();
127 
        if presented.starts_with(crate::flow::JOB_TOKEN_PREFIX) {
128 
            let job = self.block(db::flow::job_by_token(&pool, &hash, crate::flow::JOB_TOKEN_MAX_AGE_MS)).map_err(internal)?;
129 
            let Some(job) = job else { return Err(Failure(StatusCode::UNAUTHORIZED, "this job token is not valid".into())) };
130 
            let run = self.block(db::flow::run_by_id(&pool, job.run_id)).map_err(internal)?;
131 
            let Some(run) = run else { return Err(Failure(StatusCode::UNAUTHORIZED, "this job token is not valid".into())) };
132 
            return Ok(Caller::Job { repo_id: run.repo_id, also: job.also_read_ids() });
133 
        }
134 
        // An agent's grant token (keyless agents, #241): its owner's
135 
        // account, as an agent's token, writing when the grant lets its
136 
        // keys move references. Revoking the grant stops it at once.
137 
        let grant_token = presented.starts_with(crate::keyless::GRANT_TOKEN_PREFIX);
138 
        let token = if grant_token {
139 
            let grant = self.block(db::accounts::agent_grant_by_token(&pool, &hash)).map_err(internal)?;
140 
            let Some(g) = grant.filter(|g| g.live(db::now_ms())) else {
141 
                return Err(Failure(StatusCode::UNAUTHORIZED, "this agent grant is not valid (revoked, ended or unknown)".into()));
142 
            };
143 
            let _ = self.block(db::accounts::touch_agent_grant(&pool, g.id));
144 
            Some(db::accounts::Token {
145 
                id: 0,
146 
                account_id: g.account_id,
147 
                name: g.agent.clone(),
148 
                hash: String::new(),
149 
                scopes: if g.scope.split(',').any(|s| s.trim() == "refs") { "repo:read,repo:write" } else { "repo:read" }.into(),
150 
                repos: String::new(),
151 
                created_ms: g.created_ms,
152 
                expires_ms: Some(g.expires_ms),
153 
                last_used_ms: g.last_used_ms,
154 
                renew_ms: None,
155 
                agent: true,
156 
            })
157 
        } else {
158 
            self.block(db::accounts::token_by_hash(&pool, &hash)).map_err(internal)?
159 
        };
160 
        let Some(token) = token else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
161 
        if token.expired(db::now_ms()) {
162 
            return Err(Failure(StatusCode::UNAUTHORIZED, "this token has expired".into()));
163 
        }
164 
        let account = self.block(db::accounts::account_by_id(&pool, token.account_id)).map_err(internal)?;
165 
        let Some(account) = account else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
166 
        if account.banned {
167 
            return Err(Failure(StatusCode::UNAUTHORIZED, "this account is banned".into()));
168 
        }
169 
        if account.disabled {
170 
            return Err(Failure(StatusCode::UNAUTHORIZED, "this account is disabled; contact the hub's administrator".into()));
171 
        }
172 
        if !grant_token {
173 
            let _ = self.block(db::accounts::touch_token(&pool, token.id));
174 
        }
175 
        Ok(Caller::Token(Box::new(account), Box::new(token)))
176 
    }
177 
}
178
179 
fn internal(e: sqlx::Error) -> Failure {
180 
    Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
181 
}
182
183 
fn not_found() -> Failure {
184 
    Failure(StatusCode::NOT_FOUND, "no such repository".into())
185 
}
186
187 
impl Backend for HubBackend {
188 
    fn open(&self, name: &str) -> Result<Repo, Failure> {
189 
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
190 
        // `NAME.wiki` is the repository's wiki (§6.11); the row knows the
191 
        // current name, so a transferred repository still answers to its
192 
        // old one (#108).
193 
        let (parent, is_wiki) = crate::wiki::parent_of(&parsed);
194 
        let row = self.row(&parent)?;
195 
        let name = RepoName::new(&row.owner, &row.name).ok_or_else(not_found)?;
196 
        let store = if is_wiki { &self.wikis } else { &self.repos };
197 
        let repo = store.open(&name).map_err(|e| match e {
198 
            crate::repos::RepoError::NotFound => not_found(),
199 
            other => Failure(StatusCode::INTERNAL_SERVER_ERROR, other.to_string()),
200 
        })?;
201 
        Ok(match &self.acceptor {
202 
            Some(a) => repo.with_acceptor(a.0.clone(), a.1.clone()),
203 
            None => repo,
204 
        })
205 
    }
206
207 
    fn authorize(&self, name: &str, access: oxy_server::Access, headers: &HeaderMap) -> Result<(), Failure> {
208 
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
209 
        let (parent, _) = crate::wiki::parent_of(&parsed);
210 
        let row = self.row(&parent)?;
211 
        let caller = self.caller(headers)?;
212 
        if let Caller::Job { repo_id, also } = &caller {
213 
            // Only reads: of the job's own repository, and of the subrepos its
214 
            // state pins that its pusher could read when the job was claimed.
215 
            return match access {
216 
                oxy_server::Access::Read if *repo_id == row.id || also.contains(&row.id) => Ok(()),
217 
                oxy_server::Access::Read => Err(not_found()),
218 
                oxy_server::Access::Push => Err(Failure(StatusCode::FORBIDDEN, "a job token cannot push".into())),
219 
            };
220 
        }
221 
        let (role, scope_ok) = match &caller {
222 
            Caller::Job { .. } => unreachable!(),
223 
            Caller::Site => (Some(Role::Admin), true),
224 
            Caller::Anonymous => (self.block(auth::roles::role_for(&self.pool, None, &row)).map_err(internal)?, true),
225 
            Caller::Token(account, token) => {
226 
                let role = self.block(auth::roles::role_for(&self.pool, Some(account), &row)).map_err(internal)?;
227 
                let needed = match access {
228 
                    oxy_server::Access::Read => {
229 
                        token.has_scope("repo:read") || token.has_scope("repo:write") || token.has_scope("repo:admin")
230 
                    }
231 
                    oxy_server::Access::Push => token.has_scope("repo:write") || token.has_scope("repo:admin"),
232 
                };
233 
                let current = format!("{}/{}", row.owner, row.name);
234 
                let covered = token.covers(&parsed.to_string()) || token.covers(&current) || token.covers(&parent.to_string());
235 
                (role, needed && covered)
236 
            }
237 
        };
238 
        // An unreadable repository does not exist for the caller (hub spec §4.4).
239 
        if !auth::roles::allows(role, Access::Read) {
240 
            return Err(not_found());
241 
        }
242 
        // A wiki is made, empty and unbound, when someone who may write to
243 
        // its repository first reaches it, so the first page can be pushed
244 
        // as well as written on the web (§6.11); the push gives it its id.
245 
        let (_, is_wiki) = crate::wiki::parent_of(&parsed);
246 
        let may_write = match &caller {
247 
            Caller::Site => true,
248 
            Caller::Token(_, token) => token.has_scope("repo:write") || token.has_scope("repo:admin"),
249 
            _ => false,
250 
        };
251 
        if is_wiki && may_write && auth::roles::allows(role, Access::Write) && !row.archived {
252 
            if let Some(name) = RepoName::new(&row.owner, &row.name) {
253 
                if !self.wikis.exists(&name) {
254 
                    self.wikis
255 
                        .create(&name, None)
256 
                        .map_err(|e| Failure(StatusCode::INTERNAL_SERVER_ERROR, format!("make the wiki: {e}")))?;
257 
                }
258 
            }
259 
        }
260 
        match access {
261 
            oxy_server::Access::Read if scope_ok => Ok(()),
262 
            oxy_server::Access::Read => Err(Failure(StatusCode::FORBIDDEN, "this token cannot read repositories".into())),
263 
            oxy_server::Access::Push => {
264 
                if matches!(caller, Caller::Anonymous) {
265 
                    return Err(Failure(StatusCode::UNAUTHORIZED, "a personal access token is required to push".into()));
266 
                }
267 
                if !auth::roles::allows(role, Access::Write) {
268 
                    return Err(Failure(StatusCode::FORBIDDEN, "you do not have write access to this repository".into()));
269 
                }
270 
                if row.archived {
271 
                    return Err(Failure(StatusCode::FORBIDDEN, "this repository is archived and read-only".into()));
272 
                }
273 
                if !scope_ok {
274 
                    return Err(Failure(StatusCode::FORBIDDEN, "this token lacks the repo:write scope for this repository".into()));
275 
                }
276 
                Ok(())
277 
            }
278 
        }
279 
    }
280
281 
    fn default_bookmark(&self, name: &str) -> Option<String> {
282 
        let parsed = RepoName::parse(name)?;
283 
        if crate::wiki::parent_of(&parsed).1 {
284 
            return Some(crate::wiki::BOOKMARK.into());
285 
        }
286 
        self.row(&parsed).ok().map(|r| r.default_bookmark)
287 
    }
288
289 
    fn add_log(&self, name: &str, refs: &mut oxy_core::protocol::RefsResponse) {
290 
        let Some(app) = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade()) else { return };
291 
        let row = RepoName::parse(name).and_then(|p| self.row(&crate::wiki::parent_of(&p).0).ok()).map(|r| r.id);
292 
        let backfill = crate::tlog::add_log(&app, row, refs);
293 
        if !backfill.is_empty() {
294 
            self.block(crate::tlog::log(&app, backfill));
295 
        }
296 
    }
297
298 
    /// Tailor the refs answer to the client asking (oxygen#131): one that
299 
    /// does not declare `certified-tags` in `Oxy-Features` (oxy 0.5.0 and
300 
    /// older) is not sent a tag that only a certified key's certificate
301 
    /// authorizes, which it would refuse and, with it, the whole clone.
302 
    fn adjust_refs(&self, _name: &str, repo: &Repo, refs: &mut oxy_core::protocol::RefsResponse, headers: &HeaderMap) {
303 
        oxy_server::tailor_refs(repo, refs, headers);
304 
    }
305
306 
    /// Push protection (hub spec §15.4, #268): a push adding what looks
307 
    /// like a secret is refused, saying where, unless the repository turned
308 
    /// it off or a maintainer allowed that finding's fingerprint.
309 
    fn vet_push(
310 
        &self,
311 
        name: &str,
312 
        repo: &Repo,
313 
        bundle: &oxy_core::protocol::Bundle,
314 
        updates: &[oxy_core::protocol::RefUpdate],
315 
        headers: &HeaderMap,
316 
    ) -> Result<(), Failure> {
317 
        if bundle.changes.is_empty() {
318 
            return Ok(());
319 
        }
320 
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
321 
        let row = self.row(&crate::wiki::parent_of(&parsed).0)?;
322 
        if !row.push_protection {
323 
            return Ok(());
324 
        }
325 
        let scan = crate::push_protection::scan(bundle, self.scan_budget);
326 
        if let Some(note) = scan.note() {
327 
            tracing::info!("push to {name}: {note}");
328 
            self.scan_notes.lock().unwrap().insert(row.id, note);
329 
        }
330 
        if scan.findings.is_empty() {
331 
            return Ok(());
332 
        }
333 
        let allowed = self.block(db::push_protection::allowed(&self.pool, row.id)).map_err(internal)?;
334 
        let mut findings: Vec<_> = scan.findings.into_iter().filter(|f| !allowed.contains(&f.fingerprint)).collect();
335 
        if findings.is_empty() {
336 
            return Ok(());
337 
        }
338 
        crate::push_protection::locate(repo, bundle, &mut findings);
339 
        let (pusher, via) = self.pusher(headers);
340 
        let base = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade()).map(|a| a.config.base_url().to_string());
341 
        let settings =
342 
            if row.is_drop() { String::new() } else { format!("{}/{}/{}/settings", base.unwrap_or_default(), row.owner, row.name) };
343 
        let text = crate::push_protection::refusal(&findings, &settings, &via);
344 
        // Recorded for the maintainers (Settings › Push protection) and in
345 
        // the push log, as every refused push is.
346 
        let records: Vec<db::push_protection::NewFinding> = findings
347 
            .iter()
348 
            .map(|f| db::push_protection::NewFinding {
349 
                rule: &f.rule,
350 
                description: &f.description,
351 
                path: &f.path,
352 
                line: f.line as i64,
353 
                fingerprint: &f.fingerprint,
354 
                excerpt: &f.excerpt,
355 
            })
356 
            .collect();
357 
        let who = pusher.as_ref().map(|(id, n)| (*id, n.as_str()));
358 
        if let Err(e) = self.block(db::push_protection::record(&self.pool, row.id, who.map(|w| w.1).unwrap_or(""), &via, &records)) {
359 
            tracing::warn!("recording push protection findings for {name}: {e}");
360 
        }
361 
        let reason = format!(
362 
            "push protection: {} possible secret{} ({}){}",
363 
            findings.len(),
364 
            if findings.len() == 1 { "" } else { "s" },
365 
            findings.iter().take(3).map(|f| format!("{}:{}", f.path, f.line)).collect::<Vec<_>>().join(", "),
366 
            if row.is_drop() { "" } else { "; see Settings › Push protection" }
367 
        );
368 
        let refs: Vec<String> = if updates.is_empty() {
369 
            vec![oxy_core::repo::OBJECTS_RESULT.to_string()]
370 
        } else {
371 
            updates.iter().map(|u| u.name.clone()).collect()
372 
        };
373 
        let outcomes: Vec<db::repos::PushOutcome> = refs
374 
            .into_iter()
375 
            .map(|reference| db::repos::PushOutcome {
376 
                reference,
377 
                ok: false,
378 
                reason: reason.clone(),
379 
                old_state: String::new(),
380 
                new_state: String::new(),
381 
                heads: 0,
382 
                changes: 0,
383 
            })
384 
            .collect();
385 
        if let Err(e) = self.block(db::repos::record_push_via(&self.pool, row.id, who, &via, &outcomes)) {
386 
            tracing::warn!("recording refused push to {name}: {e}");
387 
        }
388 
        tracing::info!("push to {name} refused by push protection: {} finding(s)", findings.len());
389 
        Err(Failure(StatusCode::UNPROCESSABLE_ENTITY, text))
390 
    }
391
392 
    fn after_push(&self, name: &str, report: &PushReport, headers: &HeaderMap) {
393 
        let Some(parsed) = RepoName::parse(name) else { return };
394 
        // A wiki push starts no run and rings no webhook.
395 
        if crate::wiki::parent_of(&parsed).1 {
396 
            return;
397 
        }
398 
        let Ok(row) = self.row(&parsed) else { return };
399 
        let Some(parsed) = RepoName::new(&row.owner, &row.name) else { return };
400 
        let (pusher, via) = self.pusher(headers);
401 
        let pool = self.pool.clone();
402 
        // A part of a long push (objects alone) moves nothing: the push it
403 
        // belongs to is recorded when its last part names the references.
404 
        let mut results: Vec<db::repos::PushOutcome> = report
405 
            .results
406 
            .iter()
407 
            .filter(|r| r.name != oxy_core::repo::OBJECTS_RESULT)
408 
            .map(|r| {
409 
                let updated = report.updated.iter().find(|u| u.name == r.name);
410 
                db::repos::PushOutcome {
411 
                    reference: r.name.clone(),
412 
                    ok: r.ok,
413 
                    reason: r.reason.clone(),
414 
                    old_state: updated.map(|u| u.old_state.to_hex()).unwrap_or_default(),
415 
                    new_state: updated.map(|u| u.new_state.to_hex()).unwrap_or_default(),
416 
                    heads: updated.map(|u| u.heads.len() as i64).unwrap_or(0),
417 
                    changes: if r.ok { report.imported.len() as i64 } else { 0 },
418 
                }
419 
            })
420 
            .collect();
421 
        // Push protection's scan stopped early on this push or a part of it
422 
        // (#268): its accepted references say so in the log.
423 
        if !results.is_empty() {
424 
            if let Some(note) = self.scan_notes.lock().unwrap().remove(&row.id) {
425 
                for r in results.iter_mut().filter(|r| r.ok) {
426 
                    r.reason = if r.reason.is_empty() { note.clone() } else { format!("{}; {note}", r.reason) };
427 
                }
428 
            }
429 
        }
430 
        // Messages of the changes stored: `closes #N` when the default
431 
        // bookmark moved, and `#N` references on any bookmark (hub spec §8).
432 
        let bookmarks: Vec<String> = report
433 
            .updated
434 
            .iter()
435 
            .filter(|u| !u.name.starts_with("tags/") && report.results.iter().any(|r| r.ok && r.name == u.name))
436 
            .map(|u| u.name.clone())
437 
            .collect();
438 
        let imported: Vec<(String, String)> = if bookmarks.is_empty() {
439 
            vec![]
440 
        } else {
441 
            self.repos
442 
                .open(&parsed)
443 
                .map(|r| {
444 
                    report.imported.iter().filter_map(|h| r.store().get_change_meta(h).ok().map(|c| (h.to_hex(), c.message))).collect()
445 
                })
446 
                .unwrap_or_default()
447 
        };
448 
        let messages: Vec<(String, String)> = if bookmarks.contains(&row.default_bookmark) { imported.clone() } else { vec![] };
449 
        // Authors whose identity is a successor (a rotation, a renewal, a
450 
        // delegation) of an earlier one: each step kept the key pair, so
451 
        // if the chain reaches a linked key it is the same person, and the
452 
        // successor is linked too, without re-uploading it (hub spec §5.3).
453 
        let successors: Vec<(String, String, String, Vec<u8>, String)> = self
454 
            .repos
455 
            .open(&parsed)
456 
            .map(|r| {
457 
                let mut seen = std::collections::HashSet::new();
458 
                let mut out = Vec::new();
459 
                for h in &report.imported {
460 
                    let Ok(meta) = r.store().get_change_meta(h) else { continue };
461 
                    let mut key = meta.author;
462 
                    while !key.is_zero() && seen.insert(key) {
463 
                        let Ok(id) = r.store().get_identity(&key) else { break };
464 
                        let Some(prev) = id.previous else { break };
465 
                        let Ok(before) = r.store().get_identity(&prev) else { break };
466 
                        if before.ed25519_pk != id.ed25519_pk || before.mldsa65_pk != id.mldsa65_pk {
467 
                            break;
468 
                        }
469 
                        out.push((key.to_hex(), id.name.clone(), id.email.clone(), id.encode(), prev.to_hex()));
470 
                        key = prev;
471 
                    }
472 
                }
473 
                out
474 
            })
475 
            .unwrap_or_default();
476 
        let on = if bookmarks.contains(&row.default_bookmark) {
477 
            row.default_bookmark.clone()
478 
        } else {
479 
            bookmarks.first().cloned().unwrap_or_default()
480 
        };
481 
        // Bookmarks that moved, with their new heads, for OxygenFlow runs.
482 
        let moved: Vec<(String, String, Vec<oxy_core::hash::ChangeHash>)> = report
483 
            .updated
484 
            .iter()
485 
            // A bookmark sent again unchanged starts nothing.
486 
            .filter(|u| u.old_state != u.new_state && report.results.iter().any(|r| r.ok && r.name == u.name))
487 
            .map(|u| (u.name.clone(), u.new_state.to_hex(), u.heads.clone()))
488 
            .collect();
489 
        // The transparency log (#238): what the push brought, logged before
490 
        // the push is answered.
491 
        let app = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade());
492 
        if let Some(app) = app {
493 
            let to_log = self.repos.open(&parsed).map(|r| crate::tlog::push_entries(&r, Some(row.id), report)).unwrap_or_default();
494 
            if !to_log.is_empty() {
495 
                self.block(crate::tlog::log(&app, to_log));
496 
            }
497 
        }
498 
        let app_hint = self.app_hint.clone();
499 
        self.runtime.spawn(async move {
500 
            let who = pusher.as_ref().map(|(id, name)| (*id, name.as_str()));
501 
            if let Err(e) = db::repos::record_push_via(&pool, row.id, who, &via, &results).await {
502 
                tracing::warn!("recording push to {}: {e}", parsed);
503 
            }
504 
            // Oldest first: a link made for one step lets the next step link.
505 
            for (key, name, email, object, prev) in successors.iter().rev() {
506 
                let Ok(None) = db::accounts::identity_by_key(&pool, key).await else { continue };
507 
                let Ok(Some(earlier)) = db::accounts::identity_by_key(&pool, prev).await else { continue };
508 
                match db::accounts::link_identity(&pool, earlier.account_id, key, name, email, object).await {
509 
                    Ok(()) => {
510 
                        tracing::info!("linked identity {} to account {} as a successor of {}", &key[..12], earlier.account_id, &prev[..12]);
511 
                        db::accounts::audit(&pool, None, "identity.rotated", key, &format!("successor of {}", &prev[..12])).await.ok();
512 
                    }
513 
                    Err(e) => tracing::warn!("linking successor identity {}: {e}", &key[..12]),
514 
                }
515 
            }
516 
            // A drop (#279) has no issues, webhooks, mirror, Diffusion or
517 
            // runs: its push is recorded, and that is all.
518 
            if row.is_drop() {
519 
                return;
520 
            }
521 
            let app = app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade());
522 
            if let Some(app) = app {
523 
                // Issues close from every change newly on the default
524 
                // bookmark, not only those this push imported: a client's
525 
                // landing moves main to changes the hub already held.
526 
                if let Some(r) = results.iter().find(|r| r.ok && r.reference == row.default_bookmark) {
527 
                    let new_heads = moved.iter().find(|(b, _, _)| *b == row.default_bookmark).map(|(_, _, h)| h.clone()).unwrap_or_default();
528 
                    let old = r.old_state.clone();
529 
                    let old_heads = crate::repos::RepoName::new(&row.owner, &row.name);
530 
                    let old_heads = match (old_heads, oxy_core::hash::ObjectRef::from_hex(&old)) {
531 
                        (Some(n), Some(s)) => app.repos.read(&n, move |repo, _| Ok(repo.store().get_state(&s).map(|st| st.heads).unwrap_or_default())).await.unwrap_or_default(),
532 
                        _ => Vec::new(),
533 
                    };
534 
                    crate::web::issues::close_from_arrived(&app, &row, old_heads, new_heads, None).await;
535 
                } else {
536 
                    crate::web::issues::close_from_messages(&app, &row, &messages).await;
537 
                }
538 
                let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
539 
                crate::web::issues::references_from_messages(&app, &row, &imported, &on, &actor).await;
540 
                if results.iter().any(|r| r.ok && r.reference == row.default_bookmark) {
541 
                    crate::mirror::refresh(&app, &row).await;
542 
                }
543 
                let accepted: Vec<serde_json::Value> = results
544 
                    .iter()
545 
                    .filter(|r| r.ok)
546 
                    .map(|r| serde_json::json!({ "reference": r.reference, "old_state": r.old_state, "new_state": r.new_state, "heads": r.heads, "changes": r.changes }))
547 
                    .collect();
548 
                if !accepted.is_empty() {
549 
                    let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
550 
                    crate::hooks::emit(&app, &row, "push", &actor, serde_json::json!({ "updates": accepted })).await;
551 
                }
552 
                let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
553 
                // The moved bookmarks' briefs, worked out now so no page
554 
                // waits for them (#145).
555 
                if let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) {
556 
                    let all: Vec<Vec<oxy_core::hash::ChangeHash>> = moved.iter().map(|(_, _, h)| h.clone()).collect();
557 
                    let warm = app.clone();
558 
                    tokio::spawn(async move {
559 
                        warm.repos.read(&name, move |r, cache| Ok(cache.briefs(r, &all).map(|_| ())?)).await.ok();
560 
                    });
561 
                }
562 
                crate::diffusion::after_push(&app, &row, &moved).await;
563 
                // An upstream's default bookmark moved: consumers following
564 
                // every move are bumped now, those following green when
565 
                // its run passes (#148).
566 
                if let Some((_, _, heads)) = moved.iter().find(|(b, _, _)| *b == row.default_bookmark) {
567 
                    let (app, row, heads) = (app.clone(), row.clone(), heads.clone());
568 
                    tokio::spawn(async move { crate::diffusion::propagate(&app, &row, heads, false).await });
569 
                }
570 
                for (bookmark, state, heads) in &moved {
571 
                    crate::flow::on_push(&app, &row, bookmark, state, heads, &actor).await;
572 
                    crate::web::proposals::settle_after_push(&app, &row, bookmark, &actor).await;
573 
                }
574 
            }
575 
        });
576 
    }
577 
}
578
579 
/// `/{owner}/{name}/oxy/v1/{endpoint}`.
580 
pub async fn dispatch(
581 
    State(app): State<Shared>,
582 
    Path((owner, name, endpoint)): Path<(String, String, String)>,
583 
    method: Method,
584 
    headers: HeaderMap,
585 
    body: Bytes,
586 
) -> Response {
587 
    let Some(endpoint) = Endpoint::parse(&endpoint) else {
588 
        return Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response();
589 
    };
590 
    let Some(repo) = RepoName::new(&owner, &name) else {
591 
        return not_found().into_response();
592 
    };
593 
    let _guard = match endpoint {
594 
        Endpoint::Push | Endpoint::Checkpoint => Guard::Write(app.repos.write_guard(&repo).await),
595 
        _ => Guard::Read(app.repos.read_guard(&repo).await),
596 
    };
597 
    let (name, received) = (endpoint.name(), body.len() as u64);
598 
    let res = app.server.handle(repo.to_string(), endpoint, &method, &headers, body).await;
599 
    let sent = axum::body::HttpBody::size_hint(res.body()).exact().unwrap_or(0);
600 
    crate::metrics::protocol_bytes(name, received, sent);
601 
    res
602 
}
603
604 
/// `/{owner}/drops/{id}/oxy/v1/{endpoint}`: a drop is a repository to the
605 
/// wire protocol, named `owner/drops/ID` (#279).
606 
pub async fn dispatch_drop(
607 
    State(app): State<Shared>,
608 
    Path((owner, id, endpoint)): Path<(String, String, String)>,
609 
    method: Method,
610 
    headers: HeaderMap,
611 
    body: Bytes,
612 
) -> Response {
613 
    let Some(name) = RepoName::drop(&owner, &id) else {
614 
        return not_found().into_response();
615 
    };
616 
    dispatch(State(app), Path((name.owner, name.name, endpoint)), method, headers, body).await
617 
}
618
619 
enum Guard {
620 
    Read(#[allow(dead_code)] crate::repos::ReadGuard),
621 
    Write(#[allow(dead_code)] crate::repos::WriteGuard),
622 
}