use super::maps::FIREHOSE_CONNECTIONS;
use std::path::Path;
use anyhow::Context as _;
use knot_record::attachment::MAX_UPLOAD_BYTES;
use super::conf::{Block, NONE};
use super::{REVALIDATE, cached, error_body, read_limited};
use crate::blobs::UPLOAD_BLOB;
use crate::config::{ServeLimits, ValidatedConfig};
use crate::plan::repo_token;
pub(crate) fn write_locations(config: &ValidatedConfig) -> anyhow::Result<Vec<Block>> {
let cors = [
("Access-Control-Allow-Origin", "*"),
("Access-Control-Allow-Methods", "GET, POST, OPTIONS"),
(
"Access-Control-Allow-Headers",
"authorization, content-type",
),
("Access-Control-Max-Age", "86400"),
];
let location = |route: String, body_limit: &str, too_large: &str| {
let mut preflight = Block::when("$request_method", "=", "OPTIONS");
for (name, value) in cors {
preflight = preflight.directive("add_header", [name, value, "always"]);
}
let too_large_body = format!("/_ks_error_body/{too_large}.json");
let location = Block::exact(route)
.block(preflight.directive("return", ["200"]))
.block(
Block::when("$request_method", "!=", "POST")
.directive("add_header", ["Allow", "POST, OPTIONS", "always"])
.directive(
"return",
[
"405",
&error_body("InvalidRequest", "writes are POST requests"),
],
),
)
.directive("default_type", ["application/json"])
.directive("error_page", ["413", "=413", too_large_body.as_str()])
.directive("client_max_body_size", [body_limit])
.directive("client_body_buffer_size", ["64k"])
.directive("fastcgi_read_timeout", ["300s"]);
cgi_pass(throttled(location), config)
};
let upload = location(
format!("/xrpc/{UPLOAD_BLOB}"),
&MAX_UPLOAD_BYTES.to_string(),
"attachment_too_large",
);
crate::cob_write::Method::ALL
.into_iter()
.map(|method| {
location(
format!("/xrpc/{}", method.nsid()),
"64k",
"request_too_large",
)
})
.chain(
crate::repos::ALL
.into_iter()
.map(|nsid| location(format!("/xrpc/{nsid}"), "64k", "request_too_large")),
)
.chain([upload])
.collect()
}
pub(crate) fn throttled(location: Block) -> Block {
location
.directive(
"error_page",
["429", "=429", "/_ks_error_body/rate_limited.json"],
)
.directive(
"error_page",
[
"502",
"504",
"=503",
"/_ks_error_body/write_unavailable.json",
],
)
.directive("limit_req", ["zone=ks_writes", "burst=5", "nodelay"])
.directive("limit_conn", ["ks_writes_conn", "4"])
}
pub(crate) fn cgi_pass(mut location: Block, config: &ValidatedConfig) -> anyhow::Result<Block> {
let socket = config
.writes
.fastcgi_socket
.as_deref()
.context("a cgi route needs writes.fastcgi_socket")?;
let params = [
("GATEWAY_INTERFACE", "CGI/1.1".to_string()),
(
"SCRIPT_FILENAME",
config.writes.cgi_binary.display().to_string(),
),
(
"KNOT_STATIC_CONFIG",
config.config_path.display().to_string(),
),
("REQUEST_METHOD", "$request_method".into()),
("REQUEST_URI", "$request_uri".into()),
("CONTENT_TYPE", "$content_type".into()),
("CONTENT_LENGTH", "$content_length".into()),
("HTTP_AUTHORIZATION", "$http_authorization".into()),
("HTTP_CONTENT_ENCODING", "$http_content_encoding".into()),
("HTTP_GIT_PROTOCOL", "$http_git_protocol".into()),
];
for (name, value) in ¶ms {
location = location.directive("fastcgi_param", [*name, value.as_str()]);
}
Ok(location.directive("fastcgi_pass", [socket]))
}
/// `fastcgi_request_buffering` is off because buffering a push copies the pack to a temp file
pub(crate) fn push_location(config: &ValidatedConfig) -> anyhow::Result<Block> {
let location = Block::exact(PUSH_LOCATION)
.directive("internal", NONE)
.directive("fastcgi_request_buffering", ["off"])
.directive("fastcgi_read_timeout", [format!("{PUSH_READ_SECS}s")]);
cgi_pass(throttled(push_size(location)), config)
}
/// nginx checks body size in the first location a request matches, before routing to the push location
pub(crate) fn push_size(location: Block) -> Block {
location
.directive(
"error_page",
["413", "=413", "/_ks_error_body/push_too_large.json"],
)
.directive("client_max_body_size", ["8g"])
}
const PUSH_LOCATION: &str = "/_ks_push";
const PUSH_READ_SECS: u64 = 3600;
/// extra slack past `PUSH_READ_SECS` while a worker finishes an in-flight push
/// gc counts a generation retired longer ago than this as unreachable.
pub const WORKER_SHUTDOWN_SECS: u64 = PUSH_READ_SECS + 300;
pub(super) const SEND_TIMEOUT_SECS: u64 = 30;
pub(super) const INFLATE_LOCATION: &str = "/_ks_inflate";
const INFLATE_QUEUE_SECS: u32 = 30;
/// no `limit_conn` here, it would hold the one slot for the whole transfer and skip the pack's.
pub(crate) fn inflate_location(config: &ValidatedConfig) -> anyhow::Result<Block> {
let mut location = Block::exact(INFLATE_LOCATION)
.directive("internal", NONE)
.directive(
"error_page",
["429", "=429", "/_ks_error_body/fetches_saturated.json"],
)
.directive(
"error_page",
[
"502",
"504",
"=503",
"/_ks_error_body/write_unavailable.json",
],
)
.directive("client_max_body_size", ["16m"])
.directive("client_body_buffer_size", ["16m"])
.directive(
"fastcgi_read_timeout",
[format!("{}s", crate::inflate::TIMEOUT_SECS)],
)
.directive("fastcgi_param", ["KNOT_STATIC_INFLATE", "1"]);
if let Some(rate) = &config.serve.limits.inflate_rate_all {
let mut limit = vec!["zone=ks_inflate_all".to_string()];
let burst = rate.over(INFLATE_QUEUE_SECS);
if burst > 0 {
limit.push(format!("burst={burst}"));
}
location = location.directive("limit_req", limit);
}
cgi_pass(location, config)
}
pub(crate) fn firehose_location(socket: &Path) -> Block {
let refuse = |variable: &str| {
crate::atproto::reroute(Block::when(variable, "=", ""), "/_ks_error/invalid_request")
};
Block::exact("/xrpc/com.atproto.sync.subscribeRepos")
.block(refuse("$ks_fh_upgrade"))
.block(refuse("$ks_fh_key"))
.block(refuse("$ks_fh_cursor_ok"))
// args are cleared because the cursor already travels in a header.
.directive("set", ["$ks_fh_cursor", "$arg_cursor"])
.directive("set", ["$args", ""])
.directive("limit_conn", ["ks_firehose", FIREHOSE_CONNECTIONS])
.directive("proxy_pass", [format!("http://unix:{}:", socket.display())])
.directive("proxy_http_version", ["1.1"])
.directive("proxy_pass_request_headers", ["off"])
.directive("proxy_pass_request_body", ["off"])
.directive("proxy_set_header", ["Upgrade", "websocket"])
.directive("proxy_set_header", ["Connection", "upgrade"])
.directive("proxy_set_header", ["Ks-Key", "$http_sec_websocket_key"])
.directive("proxy_set_header", ["Ks-Cursor", "$ks_fh_cursor"])
.directive("proxy_buffering", ["off"])
}
pub(crate) fn limit_zones(mut http: Block, limits: &ServeLimits) -> Block {
if let Some(reads) = &limits.reads {
http = http.directive(
"limit_req_zone",
["$ks_client", "zone=ks_reads:10m", &format!("rate={reads}")],
);
}
for (count, zone) in [
(limits.read_connections, "zone=ks_reads_conn:10m"),
(limits.blob_connections, "zone=ks_blobs:10m"),
] {
if count.is_some() {
http = http.directive("limit_conn_zone", ["$ks_client", zone]);
}
}
if let Some(rate) = &limits.inflate_rate_all {
http = http.directive(
"limit_req_zone",
[
"$server_name",
"zone=ks_inflate_all:1m",
&format!("rate={rate}"),
],
);
}
if limits.packs_in_flight.is_some() {
http = http.directive("limit_conn_zone", ["$ks_pack_any", "zone=ks_packs_all:1m"]);
}
http
}
pub(super) const RETRY_AFTER_SECS: u64 = 30;
/// with a cgi, callbacks drop the code into the waiting push's state file instead of a shared port
pub(crate) fn oauth_callback(config: &ValidatedConfig) -> anyhow::Result<Block> {
let location = Block::exact("/oauth/callback");
if config.writes.fastcgi_socket.is_some() {
return cgi_pass(throttled(location), config);
}
Ok(location
.directive("proxy_pass", ["http://127.0.0.1:8918"])
.directive("proxy_set_header", ["Host", "$host"])
.directive("proxy_set_header", ["X-Real-IP", "$remote_addr"])
.directive(
"proxy_set_header",
["X-Forwarded-For", "$proxy_add_x_forwarded_for"],
)
.directive("proxy_set_header", ["X-Forwarded-Proto", "$scheme"]))
}
/// the generation id in the path stops an old worker from serving a newer generation's bytes
pub(crate) fn git_loopback(generation_path: &str, limits: &ServeLimits) -> Block {
let (generations, _) = split_generation(generation_path);
let location = Block::new(
"location",
[
"~".to_string(),
format!(
"^/_ks_git/(?<ks_git_generation>{})/(?<ks_git_token>{})$",
GENERATION.class, TOKEN.class
),
],
)
.directive("allow", ["127.0.0.1"])
.directive("deny", ["all"])
.block(
Block::when("$ks_git_file", "=", "")
.directive("return", ["400", "unsupported git request\n"]),
)
.block(
Block::when("$http_x_ks_resolve", "=", "1")
.directive(
"add_header",
[
"X-Ks-Pack",
&format!("{PACK_LOCATION}/$ks_git_generation/$ks_git_token/$ks_git_file"),
],
)
.directive("return", ["204"]),
);
pack_served(location, limits, "$ks_pack_rate").directive(
"alias",
[format!(
"{generations}/$ks_git_generation/www/_git/$ks_git_token/smart/$ks_git_file"
)],
)
}
const PACK_LOCATION: &str = "/_ks_pack";
struct Part {
class: &'static str,
takes: fn(u8) -> bool,
}
const GENERATION: Part = Part {
class: "[A-Za-z0-9_-]+",
takes: |byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-',
};
const TOKEN: Part = Part {
class: "[A-Za-z0-9_]+",
takes: |byte| byte.is_ascii_alphanumeric() || byte == b'_',
};
const FILE: Part = Part {
class: "[a-z0-9-]+",
takes: |byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-',
};
pub(crate) struct PackRedirect(String);
impl PackRedirect {
pub(crate) fn parse(path: &str) -> Option<Self> {
let rest = path.strip_prefix(PACK_LOCATION)?.strip_prefix('/')?;
let parts: Vec<&str> = rest.split('/').collect();
let fits = |text: &str, part: &Part| !text.is_empty() && text.bytes().all(part.takes);
match parts[..] {
[generation, token, file]
if fits(generation, &GENERATION) && fits(token, &TOKEN) && fits(file, &FILE) =>
{
Some(Self(path.to_string()))
}
_ => None,
}
}
pub(crate) fn as_str(&self) -> &str {
&self.0
}
}
pub(crate) fn pack_location(generation_path: &str, limits: &ServeLimits) -> Block {
let (generations, _) = split_generation(generation_path);
let location = Block::new(
"location",
[
"~".to_string(),
format!(
"^{PACK_LOCATION}/(?<ks_pack_generation>{})/(?<ks_pack_token>{})/(?<ks_pack_file>{})$",
GENERATION.class, TOKEN.class, FILE.class
),
],
)
.directive("internal", NONE);
pack_served(location, limits, "$ks_handed_rate").directive(
"alias",
[format!(
"{generations}/$ks_pack_generation/www/_git/$ks_pack_token/smart/$ks_pack_file"
)],
)
}
pub(crate) fn pack_served(mut location: Block, limits: &ServeLimits, rate: &str) -> Block {
// `max_ranges` is 0, else a Range would split the reply into slices that each skip the rate.
location = location
.directive("max_ranges", ["0"])
.block(Block::new("types", NONE));
if let Some(packs) = limits.packs_in_flight {
location = location.directive("limit_conn", ["ks_packs_all", &packs.to_string()]);
}
if limits.pack_rate.is_some() {
if let Some(after) = &limits.pack_rate_after {
location = location.directive("limit_rate_after", [after]);
}
location = location.directive("limit_rate", [rate]);
}
cached(location, "application/x-git-upload-pack-result", "no-store")
}
pub(crate) fn split_generation(generation_path: &str) -> (&str, &str) {
generation_path
.rsplit_once('/')
.expect("the served generation path is absolute")
}
pub(crate) fn repo_locations(
did: &str,
generation_path: &str,
push: bool,
inflate: &str,
limits: &ServeLimits,
) -> Vec<Block> {
let token = repo_token(did);
let mut advertisement = Block::exact(format!("/{did}/info/refs"))
.block(Block::new("limit_except", ["GET", "HEAD"]).directive("deny", ["all"]));
if push {
advertisement = advertisement.block(
Block::when("$arg_service", "=", "git-receive-pack")
.directive("rewrite", ["^", PUSH_LOCATION, "last"]),
);
}
advertisement = advertisement
.block(
Block::when("$arg_service", "!~", "^git-upload-pack$")
.directive("return", ["403", "service unsupported\n"]),
)
.block(Block::new("types", NONE));
let rate_header = match limits.pack_rate {
Some(_) => "$ks_trusted_rate",
None => "",
};
let mut locations = vec![
read_limited(
cached(
advertisement,
"application/x-git-upload-pack-advertisement",
REVALIDATE,
)
.directive(
"alias",
[format!(
"{generation_path}/www/_git/{token}/smart/$ks_git_adv"
)],
),
limits,
),
Block::exact(format!("/{did}/git-upload-pack"))
.directive("client_max_body_size", ["16m"])
.directive("client_body_buffer_size", ["16m"])
.block(
Block::when("$http_content_encoding", "~*", "^gzip$")
.directive("rewrite", ["^", inflate, "last"]),
)
.block(
Block::when("$http_content_encoding", "!~*", "^(gzip)?$")
.directive("rewrite", ["^", "/_ks_error/unsupported_encoding", "last"]),
)
.directive("proxy_http_version", ["1.1"])
// `proxy_buffering off` stops a slow client from making nginx spool the pack to disk
.directive("proxy_buffering", ["off"])
.directive("proxy_max_temp_file_size", ["0"])
.directive("proxy_method", ["GET"])
.directive("proxy_pass_request_body", ["off"])
.directive("proxy_set_header", ["Content-Length", ""])
.directive("proxy_set_header", ["X-Forwarded-For", ""])
.directive("proxy_set_header", ["X-Ks-Canned", "$ks_git_reply"])
.directive("proxy_set_header", ["X-Ks-Counted", "$ks_counted"])
.directive("proxy_set_header", ["X-Ks-Rate", rate_header])
.directive("proxy_set_header", ["X-Ks-Resolve", "$ks_resolve"])
.directive(
"proxy_pass",
[format!(
"http://127.0.0.1:$server_port/_ks_git/{}/{token}",
split_generation(generation_path).1
)],
),
];
if push {
locations.push(
push_size(Block::exact(format!("/{did}/git-receive-pack")))
.directive("rewrite", ["^", PUSH_LOCATION, "last"]),
);
}
locations
}