use std::io::{Read as _, Write as _};
use std::net::{TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::thread;
use std::time::{Duration, Instant};
use crate::config::ValidatedConfig;
use crate::manifest::{BuildStats, GenerationManifest, RepoEntry};
use crate::plan::repo_token;
mod fetch;
mod fetch_cobs;
mod firehose;
mod git_paths;
mod limits;
mod paths_fuzz;
mod social;
/// every request sends this because `real_ip_from` trusts the loopback the tests connect from.
const FORWARDED_FOR: &str = "X-Forwarded-For: 203.0.113.9";
struct Nginx {
child: Child,
}
impl Drop for Nginx {
fn drop(&mut self) {
// a killed master leaves its workers on the port. QUIT first, then TERM
signal("-QUIT", self.child.id());
let deadline = Instant::now() + Duration::from_secs(2);
while Instant::now() < deadline {
if !matches!(self.child.try_wait(), Ok(None)) {
return;
}
thread::sleep(Duration::from_millis(10));
}
signal("-TERM", self.child.id());
let _ = self.child.wait();
}
}
fn signal(signal: &str, pid: u32) -> bool {
Command::new("kill")
.args([signal, &pid.to_string()])
.stderr(Stdio::null())
.status()
.is_ok_and(|status| status.success())
}
#[test]
#[ignore = "requires nginx; set NGINX_BINARY"]
fn emitted_config_passes_nginx_test_and_serves_routes() {
let nginx_binary = PathBuf::from(
std::env::var_os("NGINX_BINARY")
.expect("NGINX_BINARY must name the real nginx binary for this ignored test"),
);
let temp = tempfile::tempdir().unwrap();
let (config, manifest, generation) = test_fixture(temp.path());
let www = generation.join("www");
std::fs::create_dir_all(www.join(".well-known")).unwrap();
std::fs::write(www.join(".well-known/did.json"), "did body").unwrap();
std::fs::write(www.join("_health.json"), "health body").unwrap();
std::fs::write(www.join("_describe.json"), "describe body").unwrap();
let index_html = crate::index_page::render(
&config.service_url().unwrap(),
&manifest.hostname,
&manifest.generation_id,
&[crate::index_page::PersonSection {
label: "ptr.pet".into(),
did: config.owner.as_str().into(),
repos: vec![crate::index_page::ServedRepo {
did: "did:plc:repo".into(),
name: "knot-static".into(),
description: Some("a static knot".into()),
active_at: None,
}],
}],
&crate::index_page::Page::new(
&config.index,
&crate::index_page::Vars {
hostname: &manifest.hostname,
owner: "ptr.pet",
owner_did: config.owner.as_str(),
knot_did: config.knot_did.as_str(),
repos: 1,
users: 1,
kaomoji: crate::index_page::kaomoji_for(&manifest.generation_id),
generation: &manifest.generation_id,
},
),
&crate::index_page::Moment::default(),
);
std::fs::write(www.join("index.html"), &index_html).unwrap();
std::fs::write(
www.join(crate::index_page::FAVICON_SVG_FILE),
crate::index_page::FAVICON_SVG,
)
.unwrap();
std::fs::write(
www.join(crate::index_page::FAVICON_PNG_FILE),
crate::index_page::FAVICON_PNG,
)
.unwrap();
std::fs::write(
www.join(crate::index_page::APPLE_TOUCH_ICON_FILE),
crate::index_page::APPLE_TOUCH_ICON,
)
.unwrap();
std::fs::write(
www.join(crate::index_page::OG_IMAGE_FILE),
crate::index_page::FAVICON_PNG,
)
.unwrap();
// a repo's cards are a set outside the generation, which it links to
let set = temp.path().join("og-set");
for (card, bytes) in [("repo.png", "repo card"), ("issues/1.png", "ticket card")] {
crate::cas::atomic_write(&set.join(card), bytes.as_bytes()).unwrap();
}
let og = www.join(crate::og_card::WWW_DIR);
std::fs::create_dir_all(&og).unwrap();
std::os::unix::fs::symlink(&set, og.join("did:plc:repo")).unwrap();
std::os::unix::fs::symlink(set.join("repo.png"), og.join("did:plc:repo.png")).unwrap();
let git_dir = www
.join("_git")
.join(repo_token("did:plc:repo"))
.join("smart");
std::fs::create_dir_all(&git_dir).unwrap();
std::fs::write(
git_dir.join("advertisement"),
b"001e# service=git-upload-pack\n0000ADVERTISED",
)
.unwrap();
std::fs::write(git_dir.join("upload-pack"), b"0008NAK\nPACKDATA").unwrap();
crate::atproto::emit(&generation, &[]).unwrap();
crate::blobs::emit(temp.path(), &generation, &[], &Default::default()).unwrap();
super::emit_at(&config, &manifest, &generation, &generation).unwrap();
let (port, _nginx) = start_nginx(&nginx_binary, &generation);
let (status, head, body) = raw_request(port, "GET / HTTP/1.1\r\n");
assert!(status.starts_with("HTTP/1.1 200 "), "index.html: {status}");
assert!(
head.to_ascii_lowercase()
.contains("content-type: text/html; charset=utf-8"),
"{head}"
);
assert!(
head.to_ascii_lowercase()
.contains("cache-control: no-cache"),
"{head}"
);
assert_eq!(body, index_html);
let (status, head, body) = raw_request(port, "HEAD / HTTP/1.1\r\n");
assert!(status.starts_with("HTTP/1.1 200 "), "HEAD /: {status}");
assert!(
head.to_ascii_lowercase()
.contains("content-type: text/html; charset=utf-8"),
"{head}"
);
assert!(body.is_empty(), "HEAD / body must be empty: {body}");
for (route, expected_type, expected_bytes) in [
(
"/favicon.svg",
"image/svg+xml",
crate::index_page::FAVICON_SVG,
),
("/favicon.ico", "image/png", crate::index_page::FAVICON_PNG),
(
"/apple-touch-icon.png",
"image/png",
crate::index_page::APPLE_TOUCH_ICON,
),
("/og.png", "image/png", crate::index_page::FAVICON_PNG),
] {
let (status, head, body) = http(port, "GET", route, &[], b"");
assert_eq!(status, 200, "{route}: {head}");
let lower = head.to_ascii_lowercase();
assert!(
lower.contains(&format!("content-type: {expected_type}")),
"{route}: {head}"
);
assert!(lower.contains("cache-control: no-cache"), "{route}: {head}");
assert_eq!(body, expected_bytes, "{route}");
}
for (route, expected) in [
("/og/did:plc:repo/issues/1.png", b"ticket card".as_slice()),
("/og/did:plc:repo/issues/2.png", b"repo card"),
("/og/did:plc:repo/commit/abc.png", b"repo card"),
("/og/did:plc:repo.png", b"repo card"),
(
"/og/did:plc:other/pulls/1.png",
crate::index_page::FAVICON_PNG,
),
("/og/did:plc:owner.png", crate::index_page::FAVICON_PNG),
] {
let (status, head, body) = http(port, "GET", route, &[], b"");
assert_eq!(status, 200, "{route}: {head}");
assert!(
head.to_ascii_lowercase()
.contains("content-type: image/png"),
"{route}: {head}"
);
assert_eq!(body, expected, "{route}");
}
let (status, head, body) = raw_request(port, "GET /not-a-route HTTP/1.1\r\n");
assert!(status.starts_with("HTTP/1.1 404 "), "{status}");
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/json"),
"{head}"
);
assert!(body.contains("not found"), "{body}");
let (status, head, body) = raw_request(port, "GET /.well-known/did.json HTTP/1.1\r\n");
assert!(status.starts_with("HTTP/1.1 200 "), "did.json: {status}");
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/did+json"),
"{head}"
);
assert_eq!(body, "did body");
let (status, _, body) = raw_request(port, "GET /xrpc/_health HTTP/1.1\r\n");
assert!(status.starts_with("HTTP/1.1 200 "), "health: {status}");
assert_eq!(body, "health body");
let (status, _, body) =
raw_request(port, "GET /xrpc/org.tangled.knot.describeKnot HTTP/1.1\r\n");
assert!(
status.starts_with("HTTP/1.1 200 "),
"describeKnot: {status}"
);
assert_eq!(body, "describe body");
let (status, head, body) = raw_request(port, "GET /xrpc/sh.tangled.owner HTTP/1.1\r\n");
assert!(status.starts_with("HTTP/1.1 404 "), "{status}");
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/json")
);
assert!(body.contains("unknown endpoint"), "{body}");
// atproto routes answer from maps. an unknown repo is a json 400, not a 404
let (status, _, body) = raw_request(
port,
"GET /xrpc/com.atproto.sync.getRepo?did=did:plc:nobody HTTP/1.1\r\n",
);
assert!(status.starts_with("HTTP/1.1 400 "), "{status}");
assert!(body.contains("RepoNotFound"), "{body}");
let (status, head, body) = raw_request(
port,
"GET /did:plc:repo/info/refs?service=git-upload-pack HTTP/1.1\r\n",
);
assert!(
status.starts_with("HTTP/1.1 200 "),
"info/refs advertisement: {status}"
);
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/x-git-upload-pack-advertisement"),
"wrong info/refs content type: {head}"
);
assert_eq!(body, "001e# service=git-upload-pack\n0000ADVERTISED");
let want = format!(
"{:04x}want {} ofs-delta\n00000009done\n",
4 + 5 + 40 + 10 + 1,
"0123456789012345678901234567890123456789"
);
let (status, head, body) = raw_request(
port,
&format!(
"POST /did:plc:repo/git-upload-pack HTTP/1.1\r\nContent-Length: {}\r\n\r\n{want}",
want.len()
),
);
assert!(
status.starts_with("HTTP/1.1 200 "),
"upload-pack POST: {status}"
);
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/x-git-upload-pack-result"),
"wrong upload-pack content type: {head}"
);
assert_eq!(body, "0008NAK\nPACKDATA");
for refused in [
"GET /did:plc:repo/info/refs HTTP/1.1\r\n",
"GET /did:plc:repo/info/refs?service=git-receive-pack HTTP/1.1\r\n",
] {
let (status, _, _) = raw_request(port, refused);
assert!(
!status.starts_with("HTTP/1.1 200 "),
"{refused} must be refused"
);
}
}
#[test]
#[ignore = "requires nginx; set NGINX_BINARY"]
fn a_dropped_nginx_leaves_no_worker_behind() {
let nginx_binary = PathBuf::from(
std::env::var_os("NGINX_BINARY")
.expect("NGINX_BINARY must name the real nginx binary for this ignored test"),
);
let temp = tempfile::tempdir().unwrap();
let (config, manifest, generation) = test_fixture(temp.path());
crate::atproto::emit(&generation, &[]).unwrap();
crate::blobs::emit(temp.path(), &generation, &[], &Default::default()).unwrap();
super::emit_at(&config, &manifest, &generation, &generation).unwrap();
let (_, nginx) = start_nginx(&nginx_binary, &generation);
let master = nginx.child.id().to_string();
let deadline = Instant::now() + Duration::from_secs(10);
let workers = loop {
let listed = Command::new("pgrep")
.args(["-P", &master])
.output()
.unwrap();
let pids: Vec<u32> = String::from_utf8_lossy(&listed.stdout)
.split_whitespace()
.map(|pid| pid.parse().unwrap())
.collect();
if !pids.is_empty() {
break pids;
}
assert!(Instant::now() < deadline, "nginx started no workers");
thread::sleep(Duration::from_millis(25));
};
drop(nginx);
let left: Vec<_> = workers
.into_iter()
.filter(|&pid| signal("-0", pid))
.collect();
assert!(left.is_empty(), "workers {left:?} outlived nginx");
}
#[test]
fn retired_workers_drain_within_the_cap_gc_waits_out() {
let temp = tempfile::tempdir().unwrap();
let (config, manifest, generation) = test_fixture(temp.path());
super::emit_at(&config, &manifest, &generation, &generation).unwrap();
let emitted = std::fs::read_to_string(generation.join("nginx/knot-static.conf")).unwrap();
let cap = crate::nginx::conf::Block::file()
.directive(
"worker_shutdown_timeout",
[format!("{}s", super::WORKER_SHUTDOWN_SECS)],
)
.to_string();
assert!(emitted.contains(cap.trim()), "{emitted}");
assert!(crate::gc::DRAIN_SECS > super::WORKER_SHUTDOWN_SECS);
}
fn start_nginx(nginx_binary: &Path, generation: &Path) -> (u16, Nginx) {
let emitted_config = generation.join("nginx/knot-static.conf");
let port = TcpListener::bind(("127.0.0.1", 0))
.unwrap()
.local_addr()
.unwrap()
.port();
let text = format!(
"pid {};\nerror_log {};\n",
generation.join("nginx.pid").display(),
generation.join("nginx/error.log").display()
) + &std::fs::read_to_string(&emitted_config)
.unwrap()
.replace("listen 8080;", &format!("listen 127.0.0.1:{port};"));
std::fs::write(&emitted_config, text).unwrap();
let check = Command::new(nginx_binary)
.args(["-t", "-c"])
.arg(&emitted_config)
.output()
.unwrap();
assert!(
check.status.success(),
"nginx -t failed:\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&check.stdout),
String::from_utf8_lossy(&check.stderr)
);
let child = Command::new(nginx_binary)
.args(["-c"])
.arg(&emitted_config)
.args(["-g", "daemon off;"])
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut nginx = Nginx { child };
let deadline = Instant::now() + Duration::from_secs(2);
loop {
if TcpStream::connect(("127.0.0.1", port)).is_ok() {
return (port, nginx);
}
if let Some(status) = nginx.child.try_wait().unwrap() {
panic!(
"nginx exited during startup with {status}; error log:\n{}",
std::fs::read_to_string(generation.join("nginx/error.log")).unwrap_or_default()
);
}
assert!(
Instant::now() < deadline,
"nginx did not start in two seconds"
);
thread::sleep(Duration::from_millis(10));
}
}
fn raw_request(port: u16, request_head: &str) -> (String, String, String) {
let mut stream = TcpStream::connect(("127.0.0.1", port)).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let (request_line, request_body) = request_head.split_once("\r\n\r\n").map_or(
(request_head.trim_end_matches("\r\n"), ""),
|(head, body)| (head.trim_end_matches("\r\n"), body),
);
write!(
stream,
"{request_line}\r\nHost: knot.test\r\n{FORWARDED_FOR}\r\nConnection: close\r\n\r\n{request_body}"
)
.unwrap();
let mut response = Vec::new();
stream.read_to_end(&mut response).unwrap();
let text = String::from_utf8_lossy(&response);
let (head, body) = text.split_once("\r\n\r\n").unwrap();
let status = head.lines().next().unwrap_or_default();
(format!("{status} "), head.to_string(), body.to_string())
}
fn repo_entry(did: &str) -> RepoEntry {
RepoEntry {
repo_did: did.into(),
owner: "did:plc:owner".into(),
rkey: "repo".into(),
name: None,
head: None,
public_refs: Vec::new(),
cob_refs: Vec::new(),
object_format: "sha1".into(),
fingerprint: "fp".into(),
git_export_hash: None,
release_id: None,
fetch_bases: Vec::new(),
shallow_branches: Vec::new(),
shallow_head: None,
moved_wants: Vec::new(),
}
}
fn test_fixture(temp: &Path) -> (ValidatedConfig, GenerationManifest, PathBuf) {
let generation = temp.join("generation");
std::fs::create_dir_all(&generation).unwrap();
let config_path = temp.join("knot-static.toml");
std::fs::write(
&config_path,
format!(
r#"
[activate]
real_ip_from = ["127.0.0.1"]
[knot]
hostname = "knot.test"
owner = "did:plc:owner"
[repo]
scan_path = "repos"
[secrets]
sealed_key_file = "sealed-keys"
[output]
root = "{}"
[atproto]
plc_directory = "https://127.0.0.1:9"
"#,
temp.join("output").display()
),
)
.unwrap();
let config = ValidatedConfig::load(&config_path).unwrap();
let manifest = GenerationManifest {
generation_id: "test".into(),
generator_profile: "test".into(),
conf_template: "test".into(),
knot_did: "did:web:knot.test".into(),
hostname: "knot.test".into(),
owner: "did:plc:owner".into(),
static_version: "test".into(),
built_by: String::new(),
profile: serde_json::json!({}),
repos: vec![repo_entry("did:plc:repo")],
rebuilt_repos: Vec::new(),
global_input_digest: None,
handles_digest: None,
stats: BuildStats::default(),
};
(config, manifest, generation)
}
fn http(
port: u16,
method: &str,
target: &str,
headers: &[(&str, &str)],
body: &[u8],
) -> (u16, String, Vec<u8>) {
let mut stream = TcpStream::connect(("127.0.0.1", port)).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
let mut head = format!(
"{method} {target} HTTP/1.1\r\nHost: knot.test\r\n{FORWARDED_FOR}\r\nConnection: close\r\n"
);
for (name, value) in headers {
head.push_str(&format!("{name}: {value}\r\n"));
}
if method == "POST" {
head.push_str(&format!("Content-Length: {}\r\n", body.len()));
}
stream.write_all(head.as_bytes()).unwrap();
stream.write_all(b"\r\n").unwrap();
stream.write_all(body).unwrap();
let mut response = Vec::new();
stream.read_to_end(&mut response).unwrap();
let split = response
.windows(4)
.position(|window| window == b"\r\n\r\n")
.expect("response head");
let head = String::from_utf8_lossy(&response[..split]).into_owned();
let status = head.split(' ').nth(1).unwrap().parse().unwrap();
(status, head, response[split + 4..].to_vec())
}
const TICKETS: [&str; 3] = ["3lubrptx57d22", "3lubrptx57d33", "3lubrptx57d44"];
const TICKET: &str = "org.tangled.track.ticket";
const VERSIONS_OVER_A_PAGE: usize = 100;
fn open_ticket() -> knot_cobs::Ticket {
knot_cobs::Ticket {
state: knot_record::ticket::TicketState::Open,
tangled_labels: None,
assignees: None,
}
}
fn assert_record_reads(
port: u16,
did: &str,
opened: &knot_types::RecordCid,
edited: &knot_types::RecordCid,
) {
let get = |target: &str| {
let (status, _, body) = http(port, "GET", target, &[], b"");
let value: serde_json::Value = serde_json::from_slice(&body)
.unwrap_or_else(|error| panic!("{target}: {status} {error}"));
(status, value)
};
let repo = did.replace(':', "%3A");
let error = |(status, body): (u16, serde_json::Value)| (status, body["error"].clone());
let rkeys = |page: &serde_json::Value| -> Vec<String> {
page["records"]
.as_array()
.unwrap()
.iter()
.map(|record| {
record["uri"]
.as_str()
.unwrap()
.rsplit('/')
.next()
.unwrap()
.into()
})
.collect()
};
let list = format!("/xrpc/com.atproto.repo.listRecords?repo={repo}&collection={TICKET}");
let newest_first: Vec<&str> = TICKETS.iter().rev().copied().collect();
let (status, all) = get(&list);
assert_eq!(status, 200);
assert_eq!(rkeys(&all), newest_first);
assert!(
all["cursor"].is_null() && all.get("cursor").is_some(),
"{all}"
);
assert_eq!(all["records"][2]["value"]["text"], "two");
let (_, backwards) = get(&format!("{list}&reverse=true"));
assert_eq!(rkeys(&backwards), TICKETS);
let (status, with_limit_3) = get(&format!("{list}&limit=3&unknown=ignored"));
assert_eq!(status, 200);
assert_eq!(rkeys(&with_limit_3), &newest_first[..1]);
for (reverse, expected) in [
("false", newest_first.as_slice()),
("true", TICKETS.as_slice()),
] {
for limit in [1, 2, 3, 50, 100] {
let mut seen = Vec::new();
let mut cursor = String::new();
loop {
let query = if cursor.is_empty() {
format!("{list}&limit={limit}&reverse={reverse}")
} else {
format!("{list}&limit={limit}&reverse={reverse}&cursor={cursor}")
};
let (status, page) = get(&query);
assert_eq!(status, 200, "{query}: {page}");
let keys = rkeys(&page);
assert!(keys.len() <= limit);
seen.extend(keys);
match page["cursor"].as_str() {
Some(next) => cursor = next.to_string(),
None => break,
}
}
assert_eq!(seen, expected, "limit={limit} reverse={reverse}");
}
let last = expected.last().unwrap();
let (status, after_last) = get(&format!("{list}&limit=50&reverse={reverse}&cursor={last}"));
assert_eq!(
(status, after_last["records"].as_array().map(Vec::len)),
(200, Some(0))
);
assert!(after_last["cursor"].is_null());
}
for bad in [
format!("{list}&cursor=nope"),
format!("{list}&limit=0"),
format!("{list}&limit=-1"),
format!("{list}&limit=101"),
format!("{list}&limit=500"),
format!("{list}&limit=abc"),
format!("{list}&reverse=maybe"),
"/xrpc/com.atproto.repo.listRecords".to_string(),
format!("/xrpc/com.atproto.repo.listRecords?collection={TICKET}"),
format!("/xrpc/com.atproto.repo.listRecords?repo={repo}"),
format!("/xrpc/com.atproto.repo.listRecords?repo={repo}&collection=not_an_nsid"),
format!(
"/xrpc/com.atproto.repo.listRecords?repo={repo}&collection=org.tangled.nothing&cursor=nope"
),
] {
assert_eq!(error(get(&bad)), (400, "InvalidRequest".into()), "{bad}");
}
let (status, none) = get(&format!(
"/xrpc/com.atproto.repo.listRecords?repo={repo}&collection=org.tangled.nothing&limit=3"
));
assert_eq!(
(status, none["records"].as_array().map(Vec::len)),
(200, Some(0))
);
assert_eq!(
error(get(&format!(
"/xrpc/com.atproto.repo.listRecords?repo=did%3Aplc%3Anope&collection={TICKET}"
))),
(400, "RepoNotFound".into())
);
let record = format!(
"/xrpc/com.atproto.repo.getRecord?repo={repo}&collection={TICKET}&rkey={}",
TICKETS[0]
);
let (status, current) = get(&format!("{record}&cid={edited}"));
assert_eq!((status, ¤t["cid"]), (200, &edited.to_string().into()));
assert_eq!(get(&record).0, 200);
for stale in [
opened.to_string(),
knot_types::RecordCid::new(knot_types::dag_cbor_cid(b"x"))
.unwrap()
.to_string(),
] {
assert_eq!(
error(get(&format!("{record}&cid={stale}"))),
(400, "RecordNotFound".into()),
"{stale}"
);
}
for missing in [
"/xrpc/com.atproto.repo.getRecord".to_string(),
format!("/xrpc/com.atproto.repo.getRecord?repo={repo}"),
format!("/xrpc/com.atproto.repo.getRecord?repo={repo}&collection={TICKET}"),
format!(
"/xrpc/com.atproto.repo.getRecord?repo={repo}&rkey={}",
TICKETS[0]
),
format!(
"/xrpc/com.atproto.repo.getRecord?collection={TICKET}&rkey={}",
TICKETS[0]
),
format!("{record}&cid=nope"),
] {
assert_eq!(
error(get(&missing)),
(400, "InvalidRequest".into()),
"{missing}"
);
}
let versions = |rkey: &str, query: &str| {
let uri = format!("at%3A%2F%2F{repo}%2F{TICKET}%2F{rkey}");
get(&format!(
"/xrpc/org.tangled.repo.listRecordVersions?uri={uri}{query}"
))
};
let (status, listed) = versions(TICKETS[0], "");
assert_eq!(status, 200, "{listed}");
assert_eq!(listed["uri"], format!("at://{did}/{TICKET}/{}", TICKETS[0]));
assert_eq!(listed["total"], 2);
assert!(listed.get("cursor").is_none());
let cids: Vec<_> = listed["versions"]
.as_array()
.unwrap()
.iter()
.map(|v| v["cid"].clone())
.collect();
assert_eq!(cids, [opened.to_string(), edited.to_string()]);
assert_eq!(listed["versions"][0]["editor"], "did:plc:alice");
assert!(
listed["versions"][0]["changeId"].is_string()
&& listed["versions"][0]["createdAt"].is_string()
);
let (_, newest) = versions(TICKETS[0], "&order=desc&cursor=0");
assert_eq!(newest["versions"][0]["cid"], edited.to_string());
let (_, literal) = get(&format!(
"/xrpc/org.tangled.repo.listRecordVersions?uri=at://{did}/{TICKET}/{}&limit=500&order=asc",
TICKETS[0]
));
assert_eq!(literal["versions"], listed["versions"]);
let (_, first_page) = versions(TICKETS[2], "&limit=100");
assert_eq!(
(
first_page["total"].as_u64(),
first_page["versions"].as_array().map(Vec::len)
),
(Some(101), Some(100))
);
let (_, last_page) = versions(
TICKETS[2],
&format!(
"&limit=100&cursor={}",
first_page["cursor"].as_str().unwrap()
),
);
assert_eq!(last_page["versions"].as_array().map(Vec::len), Some(1));
assert!(last_page.get("cursor").is_none());
let (_, one_page) = versions(TICKETS[2], "&limit=500");
assert_eq!(one_page["versions"].as_array().map(Vec::len), Some(101));
assert_eq!(
error(versions(TICKETS[0], "&limit=7")),
(400, "InvalidRequest".into())
);
assert_eq!(
error(versions(TICKETS[0], "&order=sideways")),
(400, "InvalidRequest".into())
);
assert_eq!(
error(versions(TICKETS[0], "&cursor=9")),
(400, "InvalidRequest".into())
);
assert_eq!(error(versions("nothere", "")), (404, "NotFound".into()));
let (status, elsewhere) = get(&format!(
"/xrpc/org.tangled.repo.listRecordVersions?uri=at%3A%2F%2Fdid%3Aplc%3Anope%2F{TICKET}%2F{}",
TICKETS[0]
));
assert_eq!((status, &elsewhere["error"]), (404, &"NotFound".into()));
let (status, refs) = get(&format!(
"/xrpc/org.tangled.repo.listRecordVersions?uri=at%3A%2F%2F{repo}%2Forg.tangled.git.refs%2Fself"
));
assert_eq!((status, &refs["error"]), (400, &"InvalidRequest".into()));
assert_eq!(
error(get("/xrpc/org.tangled.repo.listRecordVersions")),
(400, "InvalidRequest".into())
);
}
fn pkt(text: &str) -> Vec<u8> {
let mut out = format!("{:04x}", text.len() + 4).into_bytes();
out.extend_from_slice(text.as_bytes());
out
}
fn git_in(dir: &Path, args: &[&str]) -> String {
let output = crate::test_git()
.current_dir(dir)
.args(["-c", "user.name=T", "-c", "user.email=t@example.test"])
.args(["-c", &format!("http.extraHeader={FORWARDED_FOR}")])
.args(args)
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.output()
.unwrap();
assert!(
output.status.success(),
"git {args:?}: {}",
String::from_utf8_lossy(&output.stderr)
);
String::from_utf8(output.stdout).unwrap().trim().to_string()
}
fn contains(haystack: &[u8], needle: &[u8]) -> bool {
haystack
.windows(needle.len())
.any(|window| window == needle)
}
#[test]
#[ignore = "requires nginx; set NGINX_BINARY"]
fn smart_v2_and_atproto_routes_serve_from_files_for_sha1_and_sha256() {
use knot_types::{ObjectFormat, OwnerDid, RepoDid, RepoName, RepoRkey};
let nginx_binary = PathBuf::from(
std::env::var_os("NGINX_BINARY").expect("NGINX_BINARY must name the real nginx binary"),
);
for (format, object_format, did, hex_width) in [
(
"sha1",
ObjectFormat::SHA1,
"did:plc:abcdefghijklmnopqrstuvwx",
40,
),
(
"sha256",
ObjectFormat::SHA256,
"did:plc:bcdefghijklmnopqrstuvwxy",
64,
),
] {
let temp = tempfile::tempdir().unwrap();
let (mut config, mut manifest, generation) = test_fixture(temp.path());
config.scan_path = temp.path().join("repos");
let source = temp
.path()
.join("repos/plc")
.join(&did[8..10])
.join(&did[10..]);
std::fs::create_dir_all(&source).unwrap();
git_in(
&source,
&[
"init",
"-q",
"-b",
"main",
&format!("--object-format={format}"),
],
);
for (path, body) in [("a.txt", "one\n"), ("src/b.txt", "two\n")] {
std::fs::create_dir_all(source.join(path).parent().unwrap()).unwrap();
std::fs::write(source.join(path), body).unwrap();
git_in(&source, &["add", "-A"]);
git_in(&source, &["commit", "-q", "-m", path]);
}
git_in(&source, &["branch", "feature"]);
git_in(&source, &["tag", "-a", "v1", "-m", "v1"]);
let tip = git_in(&source, &["rev-parse", "main"]);
// a pack holding an orphan commit proves its own ref was packed, since only that ref reaches it
let tree = git_in(&source, &["rev-parse", "main^{tree}"]);
let orphan = |message: &str| git_in(&source, &["commit-tree", &tree, "-m", message]);
let cob_refs: Vec<(String, String)> = [
("refs/cobs/org.tangled.track.ticket/1", "cob"),
(
"refs/tngl/keep/did%3Aplc%3Ax/org.tangled.review.patch/1/2",
"keep",
),
]
.into_iter()
.map(|(name, message)| (name.to_string(), orphan(message)))
.collect();
for name in [
"refs/atproto/commit",
"refs/hidden/x",
"refs/cob-checkpoints/org.tangled.track.ticket/1",
] {
git_in(&source, &["update-ref", name, &orphan(name)]);
}
for (name, oid) in &cob_refs {
git_in(&source, &["update-ref", name, oid]);
}
let refs: Vec<crate::snapshot::PublicRef> = git_in(
&source,
&[
"for-each-ref",
"--format=%(objectname) %(refname)",
"refs/heads",
"refs/tags",
],
)
.lines()
.map(|line| {
let (raw, name) = line.split_once(' ').unwrap();
crate::snapshot::PublicRef {
name: name.into(),
raw: raw.into(),
peeled: None,
}
})
.collect();
assert!(refs.iter().all(|r| r.raw.len() == hex_width));
let snapshot = crate::snapshot::RepoSnapshot {
repo_did: RepoDid::new(did).unwrap(),
owner: OwnerDid::new("did:plc:owner").unwrap(),
rkey: RepoRkey::new("demo").unwrap(),
name: RepoName::new("demo").ok(),
head_capture: crate::snapshot::HeadCapture {
resolved: Some(tip.clone()),
symbolic: Some("refs/heads/main".into()),
unborn: false,
},
public_refs: refs,
cob_refs: cob_refs.clone(),
object_format,
};
let export_dir = generation.join("www/_git").join(repo_token(did));
crate::git_export::materialize_export(&config, &snapshot, &export_dir, "fp", None, None)
.unwrap();
let knot = knot_types::KnotId::new("did:web:knot.test").unwrap();
let secrets = knot_secrets::SealedStore::open(
temp.path().join("sealed.bin"),
&knot_secrets::MasterKey::new(vec![7; 32]).unwrap(),
Box::new(knot_runtime::OsEntropy),
)
.unwrap();
secrets.ensure(&knot).unwrap();
let signer = secrets.signer(&knot).unwrap();
let layout = knot_git::Layout::new(temp.path().join("cob-repos"));
let meta = temp.path().join("meta");
knot_git::Repo::create(&meta).unwrap();
let cob_repo = layout.create(&snapshot.repo_did).unwrap();
let alice = knot_types::AccountDid::new("did:plc:alice").unwrap();
let cobs = crate::test_cobs::Cobs {
repo: &cob_repo,
did: &snapshot.repo_did,
signer: &signer,
};
let (first, opened) = cobs.open_ticket(TICKETS[0], &alice, "one");
let edited = cobs.edit_ticket(first, &alice, opened, "two", 20);
cobs.open::<knot_cobs::Ticket>(TICKETS[1], &alice, 2, open_ticket(), "other", 30);
let (third, mut head) =
cobs.open::<knot_cobs::Ticket>(TICKETS[2], &alice, 3, open_ticket(), "long", 40);
for n in 0..VERSIONS_OVER_A_PAGE {
head = cobs.edit_ticket(third, &alice, head, &format!("long {n}"), 50 + n as i64);
}
let index = knot_index::Index::new(meta, layout);
let store = crate::store::RepoStore::open(&temp.path().join("repo-store"));
let settled = crate::atproto::settle(
&temp.path().join("atproto"),
&signer,
&crate::atproto::RepoSource {
did: snapshot.repo_did.clone(),
owner: snapshot.owner.clone(),
name: RepoName::new("demo").unwrap(),
export_dir: export_dir.clone(),
cobs: Some(
crate::atproto::cob_state(
&index,
&cob_repo,
&snapshot.repo_did,
&store,
&crate::datastore::Datastore::at(temp.path()),
&crate::repolock::acquire(
temp.path(),
&crate::plan::repo_key(snapshot.repo_did.as_str()),
crate::repolock::LockMode::Lifetime,
)
.unwrap(),
)
.unwrap(),
),
served: crate::store::RepoStore::served(&temp.path().join("repo")),
verify: false,
},
)
.unwrap();
crate::atproto::emit(&generation, std::slice::from_ref(&settled)).unwrap();
crate::blobs::emit(
temp.path(),
&generation,
std::slice::from_ref(&settled),
&Default::default(),
)
.unwrap();
manifest.repos = vec![repo_entry(did)];
super::emit_at(&config, &manifest, &generation, &generation).unwrap();
let (port, _nginx) = start_nginx(&nginx_binary, &generation);
let smart = |name: &str| std::fs::read(export_dir.join("smart").join(name)).unwrap();
let object_format_line = (format == "sha256").then(|| pkt("object-format=sha256\n"));
let request = |command: &str, args: &[&str]| {
let mut body = pkt(&format!("command={command}\n"));
body.extend(pkt("agent=knotmirror/0.0.1\n"));
body.extend(object_format_line.iter().flatten());
body.extend_from_slice(b"0001");
for arg in args {
body.extend(pkt(&format!("{arg}\n")));
}
body.extend_from_slice(b"0000");
body
};
let url = format!("/{did}");
let post = |body: &[u8]| http(port, "POST", &format!("{url}/git-upload-pack"), &[], body);
let (status, head, body) = http(
port,
"GET",
&format!("{url}/info/refs?service=git-upload-pack"),
&[("Git-Protocol", "version=2")],
b"",
);
assert_eq!(status, 200, "{format}: {head}");
assert!(
head.to_ascii_lowercase()
.contains("application/x-git-upload-pack-advertisement")
);
assert_eq!(body, smart("adv-v2"));
assert!(body.starts_with(b"000eversion 2\n"));
let (_, _, v0) = http(
port,
"GET",
&format!("{url}/info/refs?service=git-upload-pack"),
&[],
b"",
);
assert_eq!(v0, smart("advertisement"));
let ls_refs = request(
"ls-refs",
&[
"symrefs",
"peel",
"unborn",
"ref-prefix refs/heads/",
"ref-prefix refs/tags/",
"ref-prefix refs/cobs/",
"ref-prefix refs/tngl/keep/",
],
);
let (status, _, listed) = post(&ls_refs);
assert_eq!(status, 200, "{format}: ls-refs");
assert_eq!(listed, smart("ls-refs"));
assert!(contains(&listed, tip.as_bytes()));
assert!(contains(&listed, b"refs/heads/feature") && contains(&listed, b"refs/tags/v1"));
for (name, oid) in &cob_refs {
assert!(
contains(&listed, name.as_bytes()) && contains(&listed, oid.as_bytes()),
"{format}: ls-refs lists {name}"
);
}
let wants: Vec<String> = snapshot
.public_refs
.iter()
.map(|r| format!("want {}", r.raw))
.collect();
let mut args: Vec<&str> = vec!["thin-pack", "ofs-delta"];
args.extend(wants.iter().map(String::as_str));
let have = format!("have {}", "1".repeat(hex_width));
let mut with_haves = args.clone();
with_haves.extend([have.as_str(), "done"]);
for body in [
request("fetch", &[&args[..], &["done"]].concat()),
request("fetch", &with_haves),
] {
let (status, _, pack) = post(&body);
assert_eq!(status, 200, "{format}: fetch");
assert_eq!(pack, smart("fetch-done"));
assert!(contains(&pack, b"PACK"));
}
// long histories send request bodies past nginx's default client body buffer
let haves: Vec<String> = (0..400).map(|n| format!("have {n:0hex_width$x}")).collect();
let mut long = args.clone();
long.extend(haves.iter().map(String::as_str));
let long = request("fetch", &long);
assert!(long.len() > 16 * 1024);
let (status, _, round) = post(&long);
assert_eq!(
(status, round),
(200, smart("fetch-nodone")),
"{format}: no-done round"
);
let mut v0_want = pkt(&format!("want {tip} ofs-delta side-band-64k\n"));
v0_want.extend_from_slice(b"0000");
v0_want.extend(pkt("done\n"));
assert_eq!(post(&v0_want).2, smart("upload-pack"));
assert_eq!(post(b"garbage").0, 400);
for canned in [
"advertisement",
"adv-v2",
"ls-refs",
"ls-refs-head",
"ls-refs-head-cobs",
"upload-pack",
"fetch-done",
] {
let bytes = smart(canned);
for private in ["refs/atproto", "refs/hidden", "refs/cob-checkpoints"] {
assert!(
!contains(&bytes, private.as_bytes()),
"{canned} leaks {private}"
);
}
if !["ls-refs", "ls-refs-head-cobs"].contains(&canned) {
for cob in ["refs/cobs", "refs/tngl"] {
assert!(!contains(&bytes, cob.as_bytes()), "{canned} leaks {cob}");
}
}
}
for version in ["2", "0"] {
let clone = temp.path().join(format!("clone-v{version}"));
let status = crate::test_git()
.args([
"-c",
&format!("http.extraHeader={FORWARDED_FOR}"),
"-c",
&format!("protocol.version={version}"),
"clone",
"-q",
"--bare",
])
.arg(format!("http://127.0.0.1:{port}{url}"))
.arg(&clone)
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.output()
.unwrap();
assert!(
status.status.success(),
"{format} v{version}: {}",
String::from_utf8_lossy(&status.stderr)
);
git_in(&clone, &["fsck", "--strict"]);
let cloned = git_in(
&clone,
&[
"for-each-ref",
"--format=%(objectname) %(refname)",
"refs/heads",
"refs/tags",
],
);
let published = git_in(
&source,
&[
"for-each-ref",
"--format=%(objectname) %(refname)",
"refs/heads",
"refs/tags",
],
);
assert_eq!(cloned, published, "{format} v{version}");
}
let remote = format!("http://127.0.0.1:{port}{url}");
let fetcher = temp.path().join("clone-v2");
git_in(
&fetcher,
&[
"-c",
"protocol.version=2",
"fetch",
"-q",
&remote,
"refs/cobs/*:refs/cobs/*",
"refs/tngl/keep/*:refs/tngl/keep/*",
],
);
for (name, oid) in &cob_refs {
assert_eq!(
git_in(&fetcher, &["rev-parse", name]),
*oid,
"{format}: {name}"
);
git_in(&fetcher, &["cat-file", "-e", oid]);
}
let tracking = temp.path().join("tracking");
git_in(
temp.path(),
&[
"init",
"-q",
"--bare",
&format!("--object-format={format}"),
tracking.to_str().unwrap(),
],
);
git_in(&tracking, &["remote", "add", "origin", &remote]);
for refspec in [
"+refs/cobs/*:refs/cobs/*",
"+refs/tngl/keep/*:refs/tngl/keep/*",
] {
git_in(
&tracking,
&["config", "--add", "remote.origin.fetch", refspec],
);
}
git_in(
&tracking,
&[
"-c",
"protocol.version=2",
"-c",
"remote.origin.followRemoteHEAD=create",
"fetch",
"-q",
"origin",
],
);
for (name, oid) in &cob_refs {
assert_eq!(
git_in(&tracking, &["rev-parse", name]),
*oid,
"{format}: {name} through a remote"
);
}
let v0 = git_in(
&fetcher,
&["-c", "protocol.version=0", "ls-remote", &remote],
);
assert!(
!v0.contains("refs/cobs") && !v0.contains("refs/tngl"),
"{format}: v0 {v0}"
);
let did_query = did.replace(':', "%3A");
let (status, head, car) = http(
port,
"GET",
&format!("/xrpc/com.atproto.sync.getRepo?did={did_query}"),
&[],
b"",
);
assert_eq!(status, 200, "{format}: getRepo");
assert!(
head.to_ascii_lowercase()
.contains("application/vnd.ipld.car")
);
let car_root = knot_record::car_blocks(&car).unwrap().roots;
assert_eq!(car_root, vec![settled.commit]);
let (since_status, _, since_car) = http(
port,
"GET",
&format!("/xrpc/com.atproto.sync.getRepo?did={did_query}&since=3lubrptx57d22"),
&[],
b"",
);
assert_eq!((since_status, since_car), (200, car));
let json = |target: &str| {
let (status, _, body) = http(port, "GET", target, &[], b"");
let value: serde_json::Value = serde_json::from_slice(&body).unwrap_or_else(|error| {
panic!(
"{target}: {status} {error}: {}",
String::from_utf8_lossy(&body)
)
});
(status, value)
};
let (_, listed) = json("/xrpc/com.atproto.sync.listRepos?limit=1000");
assert_eq!(listed["repos"][0]["did"], did);
assert_eq!(listed["repos"][0]["rev"], settled.rev);
assert_eq!(json("/xrpc/com.atproto.sync.listRepos").0, 200);
assert_eq!(json("/xrpc/com.atproto.sync.listRepos?limit=7").0, 200);
assert_eq!(
json("/xrpc/com.atproto.sync.listRepos?limit=7&cursor=1").1["repos"]
.as_array()
.map(Vec::len),
Some(0)
);
for bad in [
"limit=0",
"limit=-1",
"limit=1001",
"limit=5000",
"limit=abc",
"cursor=2",
"cursor=nope",
] {
assert_eq!(
json(&format!("/xrpc/com.atproto.sync.listRepos?{bad}")).0,
400,
"{bad}"
);
}
let (_, described) = json(&format!(
"/xrpc/com.atproto.repo.describeRepo?repo={did_query}"
));
assert!(
described["collections"]
.as_array()
.unwrap()
.iter()
.any(|c| c == "org.tangled.git.refs")
);
let (_, status_body) = json(&format!("/xrpc/com.atproto.sync.getRepoStatus?did={did}"));
assert_eq!(
(status_body["active"].clone(), status_body["rev"].clone()),
(true.into(), settled.rev.clone().into())
);
let (_, latest) = json(&format!("/xrpc/com.atproto.sync.getLatestCommit?did={did}"));
assert_eq!(latest["cid"], settled.commit.to_string());
for (missing, not_found) in [
(
"/xrpc/com.atproto.sync.getRepo",
"/xrpc/com.atproto.sync.getRepo?did=did%3Aplc%3Anope",
),
(
"/xrpc/com.atproto.sync.getLatestCommit",
"/xrpc/com.atproto.sync.getLatestCommit?did=did%3Aplc%3Anope",
),
(
"/xrpc/com.atproto.sync.getRepoStatus",
"/xrpc/com.atproto.sync.getRepoStatus?did=did%3Aplc%3Anope",
),
(
"/xrpc/com.atproto.repo.describeRepo",
"/xrpc/com.atproto.repo.describeRepo?repo=did%3Aplc%3Anope",
),
] {
let (status, body) = json(missing);
assert_eq!(
(status, &body["error"]),
(400, &"InvalidRequest".into()),
"{missing}"
);
let (status, body) = json(not_found);
assert_eq!(
(status, &body["error"]),
(400, &"RepoNotFound".into()),
"{not_found}"
);
}
for bad_since in [
"nope",
"3lubrptx57d2",
"3lubrptx57d222",
"0lubrptx57d22",
"3LUBRPTX57D22",
] {
let (status, body) = json(&format!(
"/xrpc/com.atproto.sync.getRepo?did={did_query}&since={bad_since}"
));
assert_eq!(
(status, &body["error"]),
(400, &"InvalidRequest".into()),
"since={bad_since}"
);
}
let (status, record) = json(&format!(
"/xrpc/com.atproto.repo.getRecord?repo={did_query}&collection=org.tangled.repo.manifest&rkey=self"
));
assert_eq!(status, 200, "{format}: getRecord");
assert_eq!(
record["uri"],
format!("at://{did}/org.tangled.repo.manifest/self")
);
assert!(record["cid"].is_string() && record["value"]["declaration"].is_object());
assert_record_reads(port, &did, &opened, &edited);
let records = generation
.join("www/_atproto")
.join(repo_token(&did))
.join("records");
for file in std::fs::read_dir(&records).unwrap().flatten() {
std::fs::remove_file(file.path()).unwrap();
}
let (status, gone) = json(&format!(
"/xrpc/com.atproto.repo.getRecord?repo={did_query}&collection=org.tangled.repo.manifest&rkey=self"
));
assert_eq!(
(status, &gone["error"]),
(400, &"RecordNotFound".into()),
"{format}"
);
let (status, missing) = json(&format!(
"/xrpc/com.atproto.repo.getRecord?repo={did_query}&collection=org.tangled.track.ticket&rkey=1"
));
assert_eq!((status, &missing["error"]), (400, &"RecordNotFound".into()));
let (status, nobody) = json(
"/xrpc/com.atproto.repo.getRecord?repo=did%3Aplc%3Anope&collection=org.tangled.repo.manifest&rkey=self",
);
assert_eq!((status, &nobody["error"]), (400, &"RepoNotFound".into()));
let (status, unimplemented) = json(&format!(
"/xrpc/com.atproto.sync.getRecord?did={did}&collection=x&rkey=y"
));
assert_eq!(
(status, &unimplemented["error"]),
(501, &"MethodNotImplemented".into())
);
}
}
#[test]
#[ignore = "requires nginx; set NGINX_BINARY"]
fn a_map_grown_through_included_fragments_gets_a_hash_that_fits() {
let nginx_binary = PathBuf::from(
std::env::var_os("NGINX_BINARY")
.expect("NGINX_BINARY must name the real nginx binary for this ignored test"),
);
let temp = tempfile::tempdir().unwrap();
let (config, manifest, generation) = test_fixture(temp.path());
let fragments = temp.path().join("fragments");
std::fs::create_dir_all(&fragments).unwrap();
let mut includes = String::new();
let long = "p".repeat(390);
for part in 0..4 {
let rows: String = (0..6000)
.map(|i| match i % 4 {
0 => format!("k{part}x{i}{long} v;\n"),
_ => format!("k{part}x{i} v;\n"),
})
.collect();
let fragment = fragments.join(format!("{part}.conf"));
std::fs::write(&fragment, rows).unwrap();
includes += &format!(" include {};\n", fragment.display());
}
crate::git_xrpc::emit(&generation, &[]).unwrap();
let maps = generation
.join("nginx/maps")
.join(crate::git_xrpc::MAPS_FILE);
let mut text = std::fs::read_to_string(&maps).unwrap();
text += &format!("map $arg_grown $ks_grown {{\n default \"\";\n{includes}}}\n");
std::fs::write(&maps, text).unwrap();
crate::atproto::emit(&generation, &[]).unwrap();
crate::blobs::emit(temp.path(), &generation, &[], &Default::default()).unwrap();
super::emit_at(&config, &manifest, &generation, &generation).unwrap();
let emitted = std::fs::read_to_string(generation.join("nginx/knot-static.conf")).unwrap();
assert!(
emitted.contains("map_hash_max_size 32768;"),
"24000 entries need 32768"
);
let checked = generation.join("nginx/checked.conf");
let log = generation.join("nginx/error.log");
std::fs::write(
&checked,
format!(
"pid {};\nerror_log {};\n",
generation.join("nginx.pid").display(),
log.display()
) + &emitted,
)
.unwrap();
let check = Command::new(&nginx_binary)
.arg("-e")
.arg(&log)
.args(["-t", "-c"])
.arg(&checked)
.output()
.unwrap();
let said = String::from_utf8_lossy(&check.stderr);
assert!(check.status.success(), "{said}");
assert!(!said.contains("[warn]"), "{said}");
}