//! the jwt is checked before the repo lock is taken.
//! the answer goes out only after the write is published and the crawlers are told.
use std::io::Read as _;
use std::path::Path;
use std::sync::Arc;
use anyhow::Context as _;
use base64::Engine as _;
use http::StatusCode;
use jacquard_common::types::string::Nsid;
use knot_atproto::{Atproto, AtprotoError, ServiceJwt};
use knot_runtime::{ReqwestHttp, SystemClock};
use knot_types::{AccountDid, RepoDid};
use crate::blobs::{self, UPLOAD_BLOB};
use crate::cob_write::{Method, WriteError};
use crate::config::ValidatedConfig;
use crate::datastore::Datastore;
use crate::datastore::marks::{MarkName, Marked, ReplayMarks};
use crate::http_push;
use crate::inflate;
use crate::writes::{self, Actor};
pub const MAX_BODY_BYTES: usize = 64 * 1024;
const TICKET: &str = "org.tangled.track.ticket";
pub struct Request {
pub method: String,
pub target: String,
pub authorization: Option<String>,
pub content_length: Option<usize>,
pub garbled_length: bool,
pub content_type: String,
pub content_encoding: String,
pub git_protocol: Option<String>,
pub inflating: bool,
}
impl Request {
fn from_env() -> Self {
let var = |name: &str| std::env::var(name).ok();
Self {
method: var("REQUEST_METHOD").unwrap_or_default(),
target: var("REQUEST_URI").unwrap_or_default(),
authorization: var("HTTP_AUTHORIZATION"),
content_length: var("CONTENT_LENGTH").and_then(|length| length.parse().ok()),
garbled_length: var("CONTENT_LENGTH")
.is_some_and(|length| length.parse::<usize>().is_err()),
content_type: var("CONTENT_TYPE").unwrap_or_default(),
content_encoding: var("HTTP_CONTENT_ENCODING").unwrap_or_default(),
git_protocol: var("HTTP_GIT_PROTOCOL"),
inflating: var("KNOT_STATIC_INFLATE").is_some(),
}
}
pub(crate) fn path(&self) -> &str {
self.target.split(['?', '#']).next().unwrap_or_default()
}
fn method(&self) -> Result<Method, WriteError> {
self.path()
.strip_prefix("/xrpc/")
.and_then(Method::from_nsid)
.ok_or_else(|| WriteError::invalid_request("unknown endpoint"))
}
}
pub async fn run(config_path: &Path) {
let mut body = Vec::new();
let answer = match ValidatedConfig::load(config_path) {
Ok(config) => {
let config = Arc::new(config);
let request = Request::from_env();
if request.inflating {
if let Err(error) = inflate::serve(&config, &request).await {
print_error(&error, false);
}
return;
}
// a push streams its pack from stdin, and `read_body` would cap it at a write's size.
if let Some(route) = http_push::route(&request) {
print_git(http_push::serve(&config, &request, route).await);
return;
}
if request.path() == "/oauth/callback" {
print_callback(&config, &request);
return;
}
if request.path() == format!("/xrpc/{UPLOAD_BLOB}") {
upload(&config, &request, std::io::stdin().lock()).await
} else {
match read_body(&request, std::io::stdin().lock(), &mut body) {
Ok(()) => handle(&config, &request, &body).await,
Err(error) => Err(error),
}
}
}
Err(error) => Err(WriteError::internal(format!("{error:#}"))),
};
match answer {
Ok(json) => print_answer(
StatusCode::OK,
"application/json",
false,
json.to_string().as_bytes(),
),
Err(error) => print_error(&error, false),
}
}
fn print_callback(config: &ValidatedConfig, request: &Request) {
let delivered = if request.method == "GET" {
crate::oauth::deliver_code(config, &request.target)
} else {
Err(WriteError::unsupported(
StatusCode::METHOD_NOT_ALLOWED,
"InvalidRequest",
"the callback is a GET request",
))
};
match delivered {
Ok(()) => print_answer(
StatusCode::OK,
"text/html; charset=utf-8",
false,
b"<!DOCTYPE html><html><body style=\"font-family:sans-serif;padding:3rem;\"><h2>Authorization successful!</h2><p>You may return to your terminal.</p></body></html>",
),
Err(error) => print_error(&error, false),
}
}
/// the 401 carries the basic challenge, which is what makes git send credentials.
fn print_git(answer: Result<(&'static str, Vec<u8>), WriteError>) {
match answer {
Ok((content_type, body)) => print_answer(StatusCode::OK, content_type, false, &body),
Err(error) => print_error(&error, error.status() == StatusCode::UNAUTHORIZED),
}
}
pub(crate) fn print_error(error: &WriteError, challenge: bool) {
if error.status().is_server_error() {
eprintln!("knot-static cgi: {error}");
}
print_answer(
error.status(),
"application/json",
challenge,
error.response_body().to_string().as_bytes(),
);
}
fn print_answer(status: StatusCode, content_type: &str, challenge: bool, body: &[u8]) {
use std::io::Write as _;
let challenge = if challenge {
"WWW-Authenticate: Basic realm=\"knot\"\r\n"
} else {
""
};
let head = format!(
"Status: {} {}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nCache-Control: no-store\r\nAccess-Control-Allow-Origin: *\r\n{challenge}\r\n",
status.as_u16(),
status.canonical_reason().unwrap_or_default(),
body.len()
);
let mut stdout = std::io::stdout().lock();
let _ = stdout.write_all(head.as_bytes());
let _ = stdout.write_all(body);
let _ = stdout.flush();
}
pub(super) fn read_body(
request: &Request,
stdin: impl std::io::Read,
body: &mut Vec<u8>,
) -> Result<(), WriteError> {
let too_large = || {
WriteError::request_too_large(format!("a write request is at most {MAX_BODY_BYTES} bytes"))
};
if request
.content_length
.is_some_and(|length| length > MAX_BODY_BYTES)
{
return Err(too_large());
}
stdin
.take(MAX_BODY_BYTES as u64 + 1)
.read_to_end(body)
.map_err(|error| WriteError::internal(error.to_string()))?;
if body.len() > MAX_BODY_BYTES {
return Err(too_large());
}
match request.content_length {
Some(length) if length != body.len() => Err(WriteError::invalid_request(format!(
"this request's body is {} bytes, and its Content-Length says {length}",
body.len()
))),
_ => Ok(()),
}
}
pub async fn handle(
config: &Arc<ValidatedConfig>,
request: &Request,
body: &[u8],
) -> Result<serde_json::Value, WriteError> {
writable(config, request)?;
if let Some(nsid) = crate::repos::route(request) {
return crate::repos::serve(config, request, nsid, body).await;
}
let method = request.method()?;
let (actor, repo) = authenticate(config, request, method.nsid()).await?;
let published = writes::perform(config, &repo, Actor::Account(actor), method, body).await?;
if announces(method, body) {
crate::owner::announce(config, &published).await;
}
Ok(published.response())
}
fn writable(config: &ValidatedConfig, request: &Request) -> Result<(), WriteError> {
ensure_enabled(config)?;
if request.method != "POST" {
return Err(WriteError::unsupported(
StatusCode::METHOD_NOT_ALLOWED,
"InvalidRequest",
"writes are POST requests",
));
}
if request.garbled_length {
return Err(WriteError::invalid_request("Content-Length isn't a size"));
}
Ok(())
}
async fn upload(
config: &Arc<ValidatedConfig>,
request: &Request,
body: impl std::io::Read,
) -> Result<serde_json::Value, WriteError> {
writable(config, request)?;
let (actor, repo) = authenticate(config, request, UPLOAD_BLOB).await?;
writes::require_contributor(config, &repo, actor)?;
let blob = blobs::upload(
&config.output_root,
&repo,
body,
request.content_length,
&request.content_type,
)?;
Ok(serde_json::json!({ "blob": blob }))
}
fn announces(method: Method, body: &[u8]) -> bool {
let record = serde_json::from_slice::<serde_json::Value>(body).ok();
let is_pull_ticket = method == Method::CreateRecord
&& record.is_some_and(|body| {
body["collection"] == TICKET && body["record"].get("targetBranch").is_some()
});
!is_pull_ticket
}
pub(crate) fn ensure_enabled(config: &ValidatedConfig) -> Result<(), WriteError> {
config.writes.enabled.then_some(()).ok_or_else(|| {
WriteError::unsupported(
StatusCode::NOT_IMPLEMENTED,
"MethodNotImplemented",
"Method Not Implemented",
)
})
}
pub(super) fn bearer(request: &Request) -> Result<ServiceJwt, WriteError> {
request
.authorization
.as_deref()
.and_then(|value| value.split_once(' '))
.filter(|(scheme, _)| scheme.eq_ignore_ascii_case("Bearer"))
.and_then(|(_, token)| ServiceJwt::new(token.trim()).ok())
.ok_or_else(|| {
WriteError::auth_required("Missing or malformed Bearer authorization header")
})
}
pub(crate) fn verify_failed(error: AtprotoError) -> WriteError {
if error.is_transient() {
WriteError::upstream_unavailable(error.to_string())
} else {
WriteError::auth_required(error.to_string())
}
}
pub(crate) fn atproto(
config: &ValidatedConfig,
) -> Result<Atproto<ReqwestHttp, SystemClock>, WriteError> {
let transport = ReqwestHttp::new(config.http_limits())
.map_err(|error| WriteError::internal(error.to_string()))?;
config
.atproto(transport)
.map_err(|error| WriteError::internal(error.to_string()))
}
async fn authenticate(
config: &Arc<ValidatedConfig>,
request: &Request,
nsid: &str,
) -> Result<(AccountDid, RepoDid), WriteError> {
let token = bearer(request)?;
let repo = knot_atproto::token_audience(&token).map_err(|error| verify_failed(error.into()))?;
let atproto = atproto(config)?;
let nsid = Nsid::new_owned(nsid).map_err(|error| WriteError::internal(error.to_string()))?;
let actor = atproto
.verify_service_jwt_for_repo(&token, &nsid, &repo)
.await
.map_err(verify_failed)?;
mark_used(config, &actor, &token)?;
Ok((actor, repo))
}
pub(crate) async fn authenticate_addressed(
config: &ValidatedConfig,
request: &Request,
nsid: &str,
repo: Option<&RepoDid>,
) -> Result<AccountDid, WriteError> {
let token = bearer(request)?;
let atproto = atproto(config)?;
let nsid = Nsid::new_owned(nsid).map_err(|error| WriteError::internal(error.to_string()))?;
let verified = match repo {
Some(repo) if knot_atproto::token_audience(&token).is_ok_and(|aud| aud == *repo) => {
atproto
.verify_service_jwt_for_repo(&token, &nsid, repo)
.await
}
_ => atproto.verify_service_jwt(&token, &nsid).await,
};
let actor = verified.map_err(verify_failed)?;
mark_used(config, &actor, &token)?;
Ok(actor)
}
/// one write per service token, and the marks live on disk because each request is its own process.
fn mark_used(
config: &ValidatedConfig,
actor: &AccountDid,
token: &ServiceJwt,
) -> Result<(), WriteError> {
let jti = token_jti(token).ok_or_else(|| WriteError::auth_required("token carries no jti"))?;
let marked = Datastore::at(&config.output_root)
.marks::<ReplayMarks>()
.mark_once(&MarkName::of(&format!("{actor}|{jti}")))
.map_err(|error| WriteError::internal(format!("replay marks: {error:#}")))?;
match marked {
Marked::Fresh => Ok(()),
Marked::Already => Err(WriteError::auth_required(format!(
"token {jti} was already used"
))),
}
}
pub(super) fn token_jti(token: &ServiceJwt) -> Option<String> {
let payload = token.as_str().split('.').nth(1)?;
let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
.decode(payload)
.ok()?;
let claims: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
claims["jti"].as_str().map(str::to_owned)
}
/// fcgiwrap runs the binary with no arguments, which is what tells a cgi call apart.
pub fn invoked_as_cgi() -> bool {
std::env::var_os("GATEWAY_INTERFACE").is_some() && std::env::args_os().len() == 1
}
pub fn config_path_from_env() -> anyhow::Result<std::path::PathBuf> {
std::env::var_os("KNOT_STATIC_CONFIG")
.map(Into::into)
.context("a cgi run needs KNOT_STATIC_CONFIG")
}