johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! The gateway: writes on owners' behalf, for collaborators and for batches.
//!
//! The gateway is an OAuth client of its owners' PDSes. An owner delegates
//! (`dev.knotless.putDelegation`) by approving the gateway's authorization
//! request, which grants it the knotless records and blobs and nothing else
//! of the account, and the gateway keeps the session. After that, anyone on a repository's collaborator list can push
//! (`dev.knotless.pushRefs`) or commit (`dev.knotless.createCommit`) with a service
//! token from their own PDS, and the gateway writes as the owner.
//!
//! Its operator's bots (knotless-run, knotless-triage) log in through it
//! too, at `/oauth/bot`, as a client of their own asking for all of
//! knotless's scope: its session lasts as a confidential client's does,
//! and reads private repositories as an app password cannot.
//!
//! Commits to one branch queue up while a write to it is in flight. The
//! next write takes the whole queue: each commit is built on the one before
//! it, and the chain goes out as one ref update. Sixteen agents committing
//! to one branch cost about two writes, not sixteen rounds of losing a race.
use crate::{App, AppError};
use axum::Json;
use axum::body::Bytes;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::Redirect;
use knotless::commit::{self, Committed, Spec};
use knotless::forge::Target;
use knotless::gateway::{
self, AWAIT_DELEGATION, COMMIT, CommitRequest, CommitResponse, DELEGATE, PUSH, PushResults,
RefResult,
};
use knotless::landing;
use knotless::manage;
use knotless::mirror::Mirror;
use knotless::push::{self, Update};
use knotless::queue;
use knotless::records::RepoRecord;
use knotless::remote::{Remote, Snapshot};
use knotless::session::{self, Session};
use knotless::url::RepoUrl;
use knotless::{identity, oauth, org};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{oneshot, watch};
/// How many times a batch is rebuilt on a branch that moved under it.
const RETRIES: u32 = 8;
pub struct Gateway {
pub did: String,
pub endpoint: String,
/// Batches written, each one ref update however many commits it holds.
pub writes: AtomicU64,
/// Batches rebuilt because the branch moved under them.
pub rebuilt: AtomicU64,
/// Where browsers and authorization servers reach the gateway as an
/// OAuth client: the endpoint, unless that is served elsewhere too.
pub oauth_url: String,
/// How long `getDelegation` waits for an approval before it says to
/// ask again.
pub wait: Duration,
repos: Mutex<HashMap<(String, String), Lock>>,
branches: Mutex<HashMap<(String, String, String), Arc<Branch>>>,
/// Delegations waiting for their owners' approval, by state.
delegations: Mutex<HashMap<String, Delegation>>,
/// Where the bots' sessions are kept, a config directory they share:
/// none unless `--bots` names one.
pub bots: Option<PathBuf>,
/// Bot logins waiting for approval, by state.
bot_logins: Mutex<HashMap<String, (Instant, oauth::Request)>>,
/// Logins to the pages waiting for approval, by state.
pub(crate) logins: Mutex<HashMap<String, crate::login::Pending>>,
/// Logins the app finished, waiting for it to trade their code.
pub(crate) app_codes: Mutex<HashMap<String, crate::login::AppCode>>,
/// The PDS where the pages send someone making an account: one that
/// takes anyone, [`SIGN_UP_PDS`] unless `--sign-up-pds` names another.
pub sign_up_pds: String,
}
/// Where the pages send someone making an account, unless told otherwise.
pub const SIGN_UP_PDS: &str = "https://selfhosted.social";
type Outcome = watch::Sender<Option<Result<(), String>>>;
struct Delegation {
did: String,
started: Instant,
/// Taken by the browser's return.
request: Option<oauth::Request>,
done: Arc<Outcome>,
}
/// Where the gateway's OAuth client metadata is, its client ID.
pub const METADATA: &str = "/oauth-client-metadata.json";
const CALLBACK: &str = "/oauth/callback";
/// Where the bots' client metadata is, their client ID.
pub const BOT_METADATA: &str = "/oauth/bot-client-metadata.json";
const BOT_CALLBACK: &str = "/oauth/bot-callback";
/// How long a delegation waits for approval before it is dropped.
const ABANDONED: Duration = Duration::from_secs(15 * 60);
/// Held while a repository is written.
type Lock = Arc<tokio::sync::Mutex<()>>;
#[derive(Default)]
struct Branch {
queue: Mutex<Vec<Pending>>,
turn: tokio::sync::Mutex<()>,
}
struct Pending {
spec: Spec,
/// Who asked for it.
caller: String,
reply: oneshot::Sender<Result<CommitResponse, AppError>>,
}
impl Gateway {
/// A gateway reached at `endpoint`, an `https://` or `http://` URL, as
/// the `did:web` of its host.
pub fn new(endpoint: &str) -> Result<Gateway, String> {
let endpoint = endpoint.trim_end_matches('/');
let host = endpoint
.strip_prefix("https://")
.or_else(|| endpoint.strip_prefix("http://"))
.filter(|h| !h.is_empty() && !h.contains('/'))
.ok_or(format!(
"{endpoint}: a gateway URL is http(s)://host[:port]"
))?;
Ok(Gateway {
did: format!("did:web:{}", host.replace(':', "%3A")),
endpoint: endpoint.to_string(),
writes: Default::default(),
rebuilt: Default::default(),
oauth_url: endpoint.to_string(),
wait: Duration::from_secs(25),
repos: Mutex::default(),
branches: Mutex::default(),
delegations: Mutex::default(),
bots: None,
bot_logins: Mutex::default(),
logins: Mutex::default(),
app_codes: Mutex::default(),
sign_up_pds: SIGN_UP_PDS.into(),
})
}
/// The gateway as an OAuth client, signing with the key in its config
/// directory.
fn client(&self, ctx: &knotless::env::Ctx) -> Result<oauth::Client, AppError> {
self.client_at(ctx, METADATA)
}
/// The bots' client, with the same key.
fn bot_client(&self, ctx: &knotless::env::Ctx) -> Result<oauth::Client, AppError> {
self.client_at(ctx, BOT_METADATA)
}
pub(crate) fn client_at(
&self,
ctx: &knotless::env::Ctx,
metadata: &str,
) -> Result<oauth::Client, AppError> {
let key = oauth::client_key(ctx).map_err(AppError::internal)?;
Ok(oauth::Client::confidential(
&format!("{}{metadata}", self.oauth_url),
key,
))
}
}
fn forbidden(msg: impl Into<String>) -> AppError {
AppError(StatusCode::FORBIDDEN, msg.into())
}
/// A write as an owner their PDS refused: for want of scope, as their
/// grant is older than what the gateway asks for now, saying so.
fn written(err: anyhow::Error) -> AppError {
match oauth::out_of_scope(&err) {
true => forbidden(oauth::GRANT_BEHIND),
false => AppError::upstream(err),
}
}
fn bad(msg: impl Into<String>) -> AppError {
AppError(StatusCode::BAD_REQUEST, msg.into())
}
pub(crate) fn gateway(app: &App) -> Result<&Gateway, AppError> {
app.gateway
.as_ref()
.ok_or_else(|| AppError(StatusCode::NOT_FOUND, "this server is not a gateway".into()))
}
/// The DID that signed the request's service token for `lxm`.
pub(crate) async fn caller(app: &App, headers: &HeaderMap, lxm: &str) -> Result<String, AppError> {
let gw = gateway(app)?;
caller_for(app, headers, &gw.did, lxm).await
}
/// [`caller`], of a token for `aud`: this server's DID, or one of the
/// services its DID document names.
pub(crate) async fn caller_for(
app: &App,
headers: &HeaderMap,
aud: &str,
lxm: &str,
) -> Result<String, AppError> {
let token = headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.ok_or_else(|| {
AppError(
StatusCode::UNAUTHORIZED,
"a service token is required".into(),
)
})?;
knotless::auth::verify(&app.ctx, token, aud, lxm)
.await
.map_err(|e| AppError(StatusCode::UNAUTHORIZED, format!("{e:#}")))
}
pub async fn did_document(State(app): State<Arc<App>>) -> Result<Json<Value>, AppError> {
let gw = gateway(&app)?;
Ok(Json(json!({
"@context": ["https://www.w3.org/ns/did/v1"],
"id": gw.did,
"service": [{
"id": gateway::SERVICE,
"type": "KnotlessGateway",
"serviceEndpoint": gw.endpoint,
}, {
"id": knotless::moderation::LABELER,
"type": "AtprotoLabeler",
"serviceEndpoint": gw.endpoint,
}, {
"id": crate::viewer::SERVICE,
"type": "KnotlessAppView",
"serviceEndpoint": gw.endpoint,
}],
})))
}
/// The gateway's OAuth client metadata, which its client ID names.
pub async fn client_metadata(State(app): State<Arc<App>>) -> Result<Json<Value>, AppError> {
let gw = gateway(&app)?;
let client = gw.client(&app.ctx)?;
let back = format!("{}{CALLBACK}", gw.oauth_url);
Ok(Json(metadata(
&client,
"knotless gateway",
&back,
gateway::SCOPE,
)))
}
/// The bots' OAuth client metadata.
pub async fn bot_metadata(State(app): State<Arc<App>>) -> Result<Json<Value>, AppError> {
let (gw, _) = bots(&app)?;
let client = gw.bot_client(&app.ctx)?;
let back = format!("{}{BOT_CALLBACK}", gw.oauth_url);
Ok(Json(metadata(
&client,
"knotless bots",
&back,
&oauth::SCOPE,
)))
}
pub(crate) fn metadata(client: &oauth::Client, name: &str, back: &str, scope: &str) -> Value {
json!({
"client_id": client.id,
"client_name": name,
"application_type": "web",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"redirect_uris": [back],
"scope": scope,
"token_endpoint_auth_method": "private_key_jwt",
"token_endpoint_auth_signing_alg": "ES256",
"dpop_bound_access_tokens": true,
"jwks": client.jwks(),
})
}
/// The gateway, and where it keeps the bots' sessions.
fn bots(app: &App) -> Result<(&Gateway, &Path), AppError> {
let gw = gateway(app)?;
let dir = gw.bots.as_deref().ok_or_else(|| {
AppError(
StatusCode::NOT_FOUND,
"this server keeps no bots' sessions".into(),
)
})?;
Ok((gw, dir))
}
#[derive(Deserialize)]
pub struct BotLogin {
account: String,
}
/// Starts logging a bot in as `account`, one of KNOTLESS_BOT_DIDS: sends
/// the browser to its PDS, to approve as that account.
pub async fn bot_login(
State(app): State<Arc<App>>,
Query(query): Query<BotLogin>,
) -> Result<Redirect, AppError> {
let (gw, _) = bots(&app)?;
let id = identity::resolve(&app.ctx, &query.account)
.await
.map_err(AppError::upstream)?;
if !app.ctx.env.bot_dids.contains(&id.did) {
return Err(forbidden(format!(
"{} is not one of this server's bots",
id.did
)));
}
let did = id.did.clone();
let back = format!("{}{BOT_CALLBACK}", gw.oauth_url);
let client = gw.bot_client(&app.ctx)?;
let request = oauth::authorize(&app.ctx, &client, id, &did, &back, &oauth::SCOPE)
.await
.map_err(AppError::upstream)?;
let to = Redirect::to(&request.url);
let mut logins = gw.bot_logins.lock().unwrap();
logins.retain(|_, (started, _)| started.elapsed() < ABANDONED);
logins.insert(request.state.clone(), (Instant::now(), request));
Ok(to)
}
/// Where the browser comes back to from a bot's login: keeps the session,
/// and the key it refreshes with, in the bots' config directory.
pub async fn bot_callback(
State(app): State<Arc<App>>,
Query(params): Query<HashMap<String, String>>,
) -> Result<String, AppError> {
let (gw, dir) = bots(&app)?;
let state = params.get("state").map_or("", String::as_str);
let waiting = gw.bot_logins.lock().unwrap().remove(state);
let Some((_, request)) = waiting else {
return Err(bad("no bot login is waiting for this"));
};
let client = gw.bot_client(&app.ctx)?;
let redeemed = request.redeem(&app.ctx, &client, ¶ms).await;
let session = redeemed.map_err(|e| forbidden(format!("the bot was not logged in: {e:#}")))?;
// Its handle names the commits it makes, as the merge queue's.
let named = oauth::named(&app.ctx, session).await;
let session = named.map_err(AppError::upstream)?;
let mut ctx = app.ctx.clone();
ctx.env.config_dir = dir.to_path_buf();
let key = oauth::client_key(&app.ctx).map_err(AppError::internal)?;
oauth::set_client_key(&ctx, &key).map_err(AppError::internal)?;
session::save(&ctx, &session).map_err(AppError::internal)?;
crate::telemetry::said("a bot logged in", Some(&session.did));
Ok(format!(
"{} is logged in for this server's bots. You can close this tab.",
session.did
))
}
/// Starts delegating the caller's writes here: asks their PDS for a grant,
/// and answers with the page where they approve it.
pub async fn delegate(
State(app): State<Arc<App>>,
headers: HeaderMap,
) -> Result<Json<Value>, AppError> {
let did = caller(&app, &headers, DELEGATE).await?;
let gw = gateway(&app)?;
let id = identity::resolve(&app.ctx, &did)
.await
.map_err(AppError::upstream)?;
let back = format!("{}{CALLBACK}", gw.oauth_url);
let client = gw.client(&app.ctx)?;
let request = oauth::authorize(&app.ctx, &client, id, &did, &back, gateway::SCOPE)
.await
.map_err(AppError::upstream)?;
let out = json!({ "url": request.url, "state": request.state });
let mut delegations = gw.delegations.lock().unwrap();
delegations.retain(|_, d| d.started.elapsed() < ABANDONED);
delegations.insert(
request.state.clone(),
Delegation {
did,
started: Instant::now(),
request: Some(request),
done: Arc::new(watch::channel(None).0),
},
);
Ok(Json(out))
}
/// Where the owner's browser comes back to: trades the code for the
/// session the gateway writes with.
pub async fn callback(
State(app): State<Arc<App>>,
Query(params): Query<HashMap<String, String>>,
) -> Result<String, AppError> {
let gw = gateway(&app)?;
let state = params.get("state").map_or("", String::as_str);
let waiting = gw
.delegations
.lock()
.unwrap()
.get_mut(state)
.and_then(|d| Some((d.request.take()?, d.done.clone())));
let Some((request, done)) = waiting else {
return Err(bad("no delegation is waiting for this"));
};
let result = match request
.redeem(&app.ctx, &gw.client(&app.ctx)?, ¶ms)
.await
{
Ok(session) => {
app.event(
"gateway.delegated",
"",
true,
Duration::ZERO,
Some(&session.did),
);
let saved = session::save(&app.ctx, &session);
crate::spaces::in_scope_again(&session.did);
saved
}
Err(err) => Err(err),
};
let page = match &result {
Ok(()) => "The gateway now writes for you. You can close this tab.".to_string(),
Err(err) => format!("The gateway was not delegated to: {err:#}"),
};
done.send_replace(Some(result.map_err(|e| format!("{e:#}"))));
Ok(page)
}
#[derive(Deserialize)]
pub struct Awaited {
state: String,
}
/// Waits a while for the caller's delegation to be approved.
pub async fn await_delegation(
State(app): State<Arc<App>>,
headers: HeaderMap,
Query(query): Query<Awaited>,
) -> Result<Json<Value>, AppError> {
let did = caller(&app, &headers, AWAIT_DELEGATION).await?;
let gw = gateway(&app)?;
let mut done = gw
.delegations
.lock()
.unwrap()
.get(&query.state)
.filter(|d| d.did == did)
.map(|d| d.done.subscribe())
.ok_or_else(|| AppError(StatusCode::NOT_FOUND, "no such delegation".into()))?;
let waited = tokio::time::timeout(gw.wait, done.wait_for(Option::is_some)).await;
let Ok(Ok(outcome)) = waited else {
return Ok(Json(json!({ "done": false })));
};
let outcome = outcome.clone().expect("waited for an outcome");
gw.delegations.lock().unwrap().remove(&query.state);
outcome.map_err(forbidden)?;
Ok(Json(json!({ "done": true })))
}
/// Whether this server's gateway writes for `did`: it delegated writes
/// here, and the gateway keeps its session.
pub(crate) fn writes_for(app: &App, did: &str) -> bool {
app.gateway.is_some() && session::load(&app.ctx, did).is_some()
}
/// Whether the gateway writes for `did` under a grant older than what it
/// asks for now, which `did` delegating again brings up.
pub(crate) fn grant_behind(app: &App, did: &str) -> bool {
let kept = app.gateway.as_ref().and(session::load(&app.ctx, did));
kept.is_some_and(|s| oauth::behind(&s, gateway::SCOPE))
}
/// The gateway's own context, reading as `owner`, whose session it keeps
/// as they delegated: what decides a push reads a private repository's
/// space as its owner, since nobody is logged in here.
fn as_owner(app: &App, owner: &str) -> knotless::env::Ctx {
let mut ctx = app.ctx.clone();
ctx.env.identifier = Some(owner.to_string());
ctx.env.password = None;
ctx
}
/// The repository, as its owner, for a caller who may write to it.
async fn writable(
app: &App,
caller: &str,
owner: &str,
name: &str,
) -> Result<(Remote, Session, Snapshot), AppError> {
match writer(app, caller, owner, name).await? {
(_, Writer::Runner) => Err(forbidden(format!(
"{caller} is not a collaborator on knotless://{owner}/{name}"
))),
(found, Writer::Maintainer) => Ok(found),
}
}
/// Who a caller who may write to a repository is to it.
enum Writer {
/// The owner or a collaborator, who may push anything the branch rules
/// let them.
Maintainer,
/// A runner it names that is neither, which lands its merge queue and
/// pushes nothing else ([`knotless::queue::landing_refusal`]).
Runner,
}
/// The repository, as its owner, for a caller who may write to it, and who
/// they are to it.
async fn writer(
app: &App,
caller: &str,
owner: &str,
name: &str,
) -> Result<((Remote, Session, Snapshot), Writer), AppError> {
let url =
RepoUrl::parse(&format!("knotless://{owner}/{name}")).map_err(|e| bad(e.to_string()))?;
let remote = Remote::open(&app.ctx, &url)
.await
.map_err(AppError::upstream)?;
let session = session::ensure(&app.ctx, &remote.identity)
.await
.map_err(|_| forbidden(format!("{owner} has not delegated writes to this gateway")))?;
let snap = remote
.snapshot(&app.ctx)
.await
.map_err(AppError::upstream)?;
let Some(repo) = &snap.repo else {
return Err(AppError(
StatusCode::NOT_FOUND,
format!("{url}: no such repository"),
));
};
if let Some(why) = repo.value.archived() {
return Err(forbidden(why));
}
// A member of the organization the owner is pushes too, hidden or not.
let member = app.org_standing(remote.did(), caller);
let member = member.is_some_and(|s| s.standing.joined());
let named = caller == remote.did() || repo.value.is_collaborator(caller);
let writer = if named || member {
Writer::Maintainer
} else if repo.value.is_runner(caller) {
Writer::Runner
} else {
return Err(forbidden(format!(
"{caller} is not a collaborator on {url}"
)));
};
Ok(((remote, session, snap), writer))
}
async fn mirror(app: &App, remote: &Remote, snap: &Snapshot) -> Result<Mirror, AppError> {
let key = snap.key().expect("a writable repository exists");
let mirror = Mirror::open(&app.ctx, remote.did(), key)
.await
.map_err(AppError::internal)?;
mirror
.refresh(remote, snap)
.await
.map_err(AppError::upstream)?;
crate::aside(&mirror);
Ok(mirror)
}
pub async fn push(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<PushResults>, AppError> {
let started = Instant::now();
let caller = caller(&app, &headers, PUSH).await?;
let (header, pack) = gateway::split_push(&body).map_err(|e| bad(format!("{e:#}")))?;
let mut updates = Vec::new();
for u in &header.updates {
if u.src.as_deref().is_some_and(|s| !gateway::is_sha(s)) || !u.dst.starts_with("refs/") {
return Err(bad(format!("bad update of {}", u.dst)));
}
updates.push(Update {
src: u.src.clone(),
dst: u.dst.clone(),
force: u.force,
});
}
let lock = gateway(&app)?
.repos
.lock()
.unwrap()
.entry((header.repo.clone(), header.name.clone()))
.or_default()
.clone();
let _turn = lock.lock().await;
let found = writer(&app, &caller, &header.repo, &header.name).await?;
let ((remote, session, snap), writer) = found;
let mirror = mirror(&app, &remote, &snap).await?;
if !pack.is_empty() {
mirror
.git
.index_pack(pack)
.await
.map_err(|e| bad(format!("{e:#}")))?;
}
let tips: Vec<String> = updates.iter().filter_map(|u| u.src.clone()).collect();
if let Some(sha) = mirror
.git
.missing(&tips)
.await
.map_err(AppError::internal)?
.into_iter()
.next()
{
return Err(bad(format!("the pack does not hold {sha}")));
}
// Protected branches hold whoever pushes here; the owner can write
// past the gateway. A collaborator's merge is held to the merge gate,
// read as the repository is now: as `writer` read it, under this
// push's turn, so it is not read again, for the gate or a runner's
// landing. Read as the owner, whose space a private one is in.
let ctx = as_owner(&app, remote.did());
let record = snap.repo.as_ref().map(|r| &r.value);
let reviewed = |r: &RepoRecord| r.rules.iter().any(|x| x.reviews.is_some());
let gated = record.is_some_and(|r| !r.runners.is_empty() || reviewed(r));
let gated = gated && caller != remote.did();
let runner = matches!(writer, Writer::Runner);
let opened = match runner || gated {
true => Some(Target::read(&ctx, remote.clone(), snap.clone()).await),
false => None,
};
let opened = opened.transpose().map_err(AppError::upstream)?;
if let (true, Some(target)) = (runner, &opened) {
let refusal = queue::landing_refusal(&ctx, target, &mirror.git, &updates).await;
if let Some(why) = refusal.map_err(AppError::internal)? {
return Err(forbidden(format!(
"{caller} runs {}'s checks, and lands its merge queue and pulls set to merge when their checks pass here, nothing else: {why}",
remote.url
)));
}
}
let target = opened.filter(|_| gated);
let (mut allowed, mut refused) = (Vec::new(), Vec::new());
for update in updates {
let refusal = gateway::refusal(&mirror.git, &snap, &update, remote.did(), &caller).await;
let mut refusal = refusal.map_err(AppError::internal)?;
if let (None, Some(t)) = (&refusal, &target) {
let held = landing::gate_refusal(&ctx, t, &mirror.git, &update, &caller).await;
refusal = held.map_err(AppError::upstream)?;
}
match refusal {
Some(reason) => refused.push((update.dst, Err(reason))),
None => allowed.push(update),
}
}
let updates = allowed;
let mut report = push::push_by(
&app.ctx,
&remote,
&snap,
&mirror.git,
&session.auth(),
&updates,
Some(&caller),
)
.await
.map_err(written)?;
report.extend(refused);
app.event("gateway.push", "", true, started.elapsed(), Some(&caller));
// By the owner it wrote for, whoever asked: what delegating led to,
// once a ref moved.
let moved = report.iter().any(|(_, res)| res.is_ok());
if moved {
let owner = Some(remote.did());
app.event("gateway.served", "", true, started.elapsed(), owner);
}
if moved && remote.space.is_some() && !pack.is_empty() {
seed(&app, &caller, remote.did(), &snap, body.slice_ref(pack));
}
Ok(Json(PushResults {
results: report
.into_iter()
.map(|(name, res)| RefResult {
name,
error: res.err(),
})
.collect(),
}))
}
/// Puts `pack`, as `caller` pushed it to `did`'s private repository
/// `snap`, in the mirror of it the pages keep for them, if they keep one:
/// aside, as the push is answered. Their next look then reads its records
/// and not this pack again from the PDS. Only what they sent goes in,
/// theirs to read already, so their mirror is still theirs alone
/// (`docs/design/access.md`). A pack whose bases that mirror lacks is not
/// taken in, and that look fetches it as before.
fn seed(app: &App, caller: &str, did: &str, snap: &Snapshot, pack: Bytes) {
let key = snap.key().expect("a writable repository exists");
let ctx = crate::login::ctx_for(app, caller);
let dir = knotless::mirror::path(&ctx, did, key);
if dir.join("HEAD").exists() {
tokio::spawn(async move { knotless::git::Git::new(dir).index_pack(&pack).await.ok() });
}
}
pub async fn commit(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(req): Json<CommitRequest>,
) -> Result<Json<CommitResponse>, AppError> {
let started = Instant::now();
let caller = caller(&app, &headers, COMMIT).await?;
if req.branch.is_empty() || req.message.is_empty() {
return Err(bad("a commit needs a branch and a message"));
}
// Checked before queueing, so a batch holds only commits that may land.
let (remote, _, snap) = writable(&app, &caller, &req.repo, &req.name).await?;
let record = snap.repo.as_ref().map(|r| &r.value);
if record.is_some_and(|r| r.held_to_queue(remote.did(), &caller, &req.branch)) {
return Err(forbidden(format!(
"{} takes pulls only through the merge queue: commit to another branch and open a pull",
req.branch
)));
}
if record.is_some_and(|r| r.held_to_reviews(remote.did(), &caller, &req.branch)) {
return Err(forbidden(format!(
"{} takes changes only through reviewed pulls: commit to another branch and open a pull",
req.branch
)));
}
let branch = gateway(&app)?
.branches
.lock()
.unwrap()
.entry((req.repo.clone(), req.name.clone(), req.branch.clone()))
.or_default()
.clone();
let (reply, answer) = oneshot::channel();
branch.queue.lock().unwrap().push(Pending {
spec: req.spec(),
caller: caller.clone(),
reply,
});
{
let _turn = branch.turn.lock().await;
let batch = std::mem::take(&mut *branch.queue.lock().unwrap());
// Empty when the write before this one took this commit along.
if !batch.is_empty() {
land(&app, &caller, &req, batch).await;
}
}
let answered = answer.await.expect("every queued commit is answered");
let ok = answered.is_ok();
app.event("gateway.commit", "", ok, started.elapsed(), Some(&caller));
if ok {
let owner = Some(remote.did());
app.event("gateway.served", "", true, started.elapsed(), owner);
}
answered.map(Json)
}
/// Builds the batch on the branch and writes it, rebuilding on a branch
/// that moved, and answers every commit in it.
async fn land(app: &App, caller: &str, req: &CommitRequest, batch: Vec<Pending>) {
match attempt(app, caller, req, &batch).await {
Ok(results) => {
for (pending, result) in batch.into_iter().zip(results) {
let _ = pending.reply.send(result);
}
}
Err(err) => {
for pending in batch {
let _ = pending.reply.send(Err(AppError(err.0, err.1.clone())));
}
}
}
}
type Answers = Vec<Result<CommitResponse, AppError>>;
/// Who moves the branch for a batch: its one caller, or no one but the
/// owner when it gathered several.
fn pusher(batch: &[Pending]) -> Option<&str> {
let first = batch.first()?.caller.as_str();
batch.iter().all(|p| p.caller == first).then_some(first)
}
async fn attempt(
app: &App,
caller: &str,
req: &CommitRequest,
batch: &[Pending],
) -> Result<Answers, AppError> {
let refname = format!("refs/heads/{}", req.branch);
let mut round = 0;
loop {
let (remote, session, snap) = writable(app, caller, &req.repo, &req.name).await?;
let mirror = mirror(app, &remote, &snap).await?;
let start = commit::start(&snap, &batch[0].spec);
let mut tip = start.clone();
let mut built: Vec<Result<Committed, String>> = Vec::new();
for pending in batch {
if let Err(err) = commit::conflicts(&mirror.git, &pending.spec, tip.as_deref()).await {
built.push(Err(format!("{err:#}")));
continue;
}
// An edit that no longer finds its text fails its own commit,
// not the batch.
let spec = match commit::resolve(&mirror.git, tip.as_deref(), &pending.spec).await {
Ok(spec) => spec,
Err(err) => {
built.push(Err(format!("{err:#}")));
continue;
}
};
let sha = commit::build(&app.ctx, &mirror.git, tip.as_deref(), &spec, remote.did())
.await
.map_err(AppError::input)?;
let parent = tip.clone();
if let Some(sha) = &sha {
tip = Some(sha.clone());
}
built.push(Ok(Committed {
sha: tip.clone().unwrap_or_default(),
parent,
attempts: 1,
changed: sha.is_some(),
}));
}
let landed = built.iter().flatten().filter(|c| c.changed).count();
if tip != start {
let update = Update {
src: tip.clone(),
dst: refname.clone(),
force: true,
};
// What a push of this tip would be refused, a commit is: a new
// branch under a linear rule takes no merge commits from its base.
let owner = remote.did();
let held = gateway::refusal(&mirror.git, &snap, &update, owner, caller).await;
if let Some(why) = held.map_err(AppError::internal)? {
return Err(forbidden(format!("{refname}: {why}")));
}
let report = push::push_by(
&app.ctx,
&remote,
&snap,
&mirror.git,
&session.auth(),
&[update],
pusher(batch),
)
.await
.map_err(written)?;
gateway(app)?.writes.fetch_add(1, Ordering::Relaxed);
// Forced against the generation read above, so the one refusal
// is that the branch moved: rebuild on where it went.
if let Err(reason) = &report[0].1 {
let gave_up = AppError(StatusCode::CONFLICT, format!("{refname}: {reason}"));
(round < RETRIES).then_some(()).ok_or(gave_up)?;
gateway(app)?.rebuilt.fetch_add(1, Ordering::Relaxed);
round += 1;
continue;
}
}
let answers = built
.into_iter()
.map(|c| match c {
Ok(c) => Ok(CommitResponse {
sha: c.sha,
parent: c.parent,
changed: c.changed,
batched: landed,
}),
Err(msg) => Err(AppError(StatusCode::CONFLICT, msg)),
})
.collect();
return Ok(answers);
}
}
/// A call to an organization's gateway naming only the organization.
#[derive(Deserialize)]
pub struct OrgCall {
org: String,
}
/// A member's call to be taken in, saying which of their join records
/// counts, if they keep both.
#[derive(Deserialize)]
pub struct JoinCall {
org: String,
visibility: Option<org::Visibility>,
}
/// An admin's call to an organization's gateway about one account.
#[derive(Deserialize)]
pub struct MemberCall {
org: String,
member: String,
role: Option<String>,
}
/// The organization `org`'s session, which this gateway keeps.
pub(crate) async fn organization(app: &App, org: &str) -> Result<Session, AppError> {
gateway(app)?;
let id = identity::resolve(&app.ctx, org)
.await
.map_err(AppError::upstream)?;
session::ensure(&app.ctx, &id)
.await
.map_err(|_| forbidden(format!("{org} has not delegated writes to this gateway")))
}
/// The organization `org`'s session, for `caller`, who must be the
/// organization or one of its admins who joined.
async fn administered(app: &App, caller: &str, org: &str) -> Result<Session, AppError> {
let session = organization(app, org).await?;
match app.org_admits(&session.did, caller) {
true => Ok(session),
false => Err(forbidden(format!("{caller} is not an admin of {org}"))),
}
}
/// Runs `change` to the organization `session` is, about `member`, alone
/// among changes to it, then writes what it changed into the indexes.
async fn changing<T>(
app: &App,
session: &Session,
member: &str,
change: impl std::future::Future<Output = anyhow::Result<T>>,
) -> Result<Json<T>, AppError> {
let lock = app.org_lock(&session.did);
let _held = lock.lock().await;
let done = change.await;
app.reindex_org(&session.did, member).await;
done.map(Json).map_err(unchanged)
}
/// The account `member` names, by DID, as the organization's records name
/// it.
async fn account(app: &App, member: &str) -> Result<String, AppError> {
let found = identity::resolve(&app.ctx, member).await;
found.map(|id| id.did).map_err(|e| bad(format!("{e:#}")))
}
/// A change to an organization the gateway was refused, or refused itself.
fn unchanged(err: anyhow::Error) -> AppError {
match oauth::out_of_scope(&err) {
true => forbidden(oauth::GRANT_BEHIND),
false => AppError(StatusCode::CONFLICT, format!("{err:#}")),
}
}
pub async fn put_org_member(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(call): Json<MemberCall>,
) -> Result<Json<org::Invited>, AppError> {
let caller = caller(&app, &headers, gateway::PUT_ORG_MEMBER).await?;
let session = administered(&app, &caller, &call.org).await?;
let member = account(&app, &call.member).await?;
let role = org::Role::of(call.role.as_deref());
let gateway = gateway(&app)?;
let done = org::invite_as(&app.ctx, &session, &member, role, &gateway.did);
changing(&app, &session, &member, done).await
}
pub async fn remove_org_member(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(call): Json<MemberCall>,
) -> Result<Json<org::Removed>, AppError> {
let caller = caller(&app, &headers, gateway::REMOVE_ORG_MEMBER).await?;
let session = administered(&app, &caller, &call.org).await?;
let member = account(&app, &call.member).await?;
let done = org::remove_as(&app.ctx, &session, &member);
changing(&app, &session, &member, done).await
}
/// Takes the caller in, once they wrote their join record.
pub async fn join_org(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(call): Json<JoinCall>,
) -> Result<Json<org::Joined>, AppError> {
let caller = caller(&app, &headers, gateway::JOIN_ORG).await?;
let session = organization(&app, &call.org).await?;
let done = org::admit_as(&app.ctx, &session, &caller, call.visibility);
changing(&app, &session, &caller, done).await
}
/// Takes the caller out: they ended the membership as they took back their
/// join record, and this takes back what it let them read.
pub async fn leave_org(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(call): Json<OrgCall>,
) -> Result<Json<org::Removed>, AppError> {
let caller = caller(&app, &headers, gateway::LEAVE_ORG).await?;
let session = organization(&app, &call.org).await?;
let done = org::remove_as(&app.ctx, &session, &caller);
changing(&app, &session, &caller, done).await
}
/// An admin's call to make a repository as an organization.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OrgRepoCall {
org: String,
name: String,
#[serde(default)]
private: bool,
description: Option<String>,
default_branch: Option<String>,
runners: Option<Vec<org::Account>>,
}
/// Makes a repository as the organization, for one of its admins.
pub async fn create_org_repo(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(call): Json<OrgRepoCall>,
) -> Result<Json<org::Made>, AppError> {
let caller = caller(&app, &headers, gateway::CREATE_ORG_REPO).await?;
let session = administered(&app, &caller, &call.org).await?;
// Made alone, as its members are added: one taken in or out meanwhile
// would be left out of it, or in.
let lock = app.org_lock(&session.did);
let _held = lock.lock().await;
let url = RepoUrl {
authority: call.org,
name: call.name,
};
let new = manage::NewRepo {
private: call.private,
description: call.description,
default_branch: call.default_branch,
runners: call.runners.map(|r| r.into_iter().map(|a| a.did).collect()),
..Default::default()
};
let made = manage::create(&app.ctx, &url, &new).await;
made.map(|m| Json(m.into())).map_err(unchanged)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn writes_refused_for_want_of_scope_say_to_delegate_again() {
let refused = || anyhow::anyhow!("InvalidToken (403): Scope does not permit create");
for answered in [written(refused()), unchanged(refused())] {
assert_eq!(answered.0, StatusCode::FORBIDDEN);
assert_eq!(answered.1, oauth::GRANT_BEHIND);
}
assert_eq!(written(anyhow::anyhow!("down")).0, StatusCode::BAD_GATEWAY);
assert_eq!(unchanged(anyhow::anyhow!("no")).0, StatusCode::CONFLICT);
}
}