use std::io::{BufRead as _, BufReader, Read as _, Write as _};
use std::net::TcpStream;
use std::os::unix::net::UnixListener;
use std::path::PathBuf;
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
use super::{FORWARDED_FOR, start_nginx, test_fixture};
const KEY: &str = "dGhlIHNhbXBsZSBub25jZQ==";
fn upgrade(port: u16, method: &str, query: &str, headers: &str) -> String {
let mut stream = TcpStream::connect(("127.0.0.1", port)).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
write!(
stream,
"{method} /xrpc/com.atproto.sync.subscribeRepos{query} HTTP/1.1\r\nHost: knot.test\r\n\
{FORWARDED_FOR}\r\n{headers}\r\n"
)
.unwrap();
let mut response = Vec::new();
let _ = stream.read_to_end(&mut response);
String::from_utf8_lossy(&response).into_owned()
}
fn websocket(key: &str) -> String {
format!(
"Upgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {key}\r\n\
Sec-WebSocket-Version: 13\r\nCookie: secret=1\r\nX-Junk: yes\r\n"
)
}
#[test]
#[ignore = "requires nginx; set NGINX_BINARY"]
fn subscribe_repos_reaches_the_server_as_only_the_checked_key_and_cursor() {
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 (mut config, manifest, generation) = test_fixture(temp.path());
let socket = temp.path().join("fh.sock");
config.firehose_socket = Some(socket.clone());
crate::atproto::emit(&generation, &[]).unwrap();
crate::blobs::emit(temp.path(), &generation, &[], &Default::default()).unwrap();
super::super::emit_at(&config, &manifest, &generation, &generation).unwrap();
let (port, _nginx) = start_nginx(&nginx_binary, &generation);
let listener = UnixListener::bind(&socket).unwrap();
let (heads, received) = mpsc::channel();
thread::spawn(move || {
for stream in listener.incoming() {
let mut stream = stream.unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut head = String::new();
loop {
let mut line = String::new();
if reader.read_line(&mut line).unwrap() == 0 || line == "\r\n" {
break;
}
head.push_str(&line);
}
let _ = stream.write_all(
b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n\
Connection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n\
\x82\x02hi",
);
heads.send(head).unwrap();
}
});
let reply = upgrade(port, "GET", "?cursor=42", &websocket(KEY));
assert!(reply.starts_with("HTTP/1.1 101 "), "{reply}");
assert!(
reply.ends_with("\u{fffd}\u{2}hi"),
"the frame went through: {reply:?}"
);
let head = received.recv_timeout(Duration::from_secs(5)).unwrap();
let mut lines = head.lines();
assert_eq!(
lines.next(),
Some("GET /xrpc/com.atproto.sync.subscribeRepos HTTP/1.1")
);
let headers: Vec<String> = lines.map(str::to_ascii_lowercase).collect();
for wanted in [
format!("ks-key: {}", KEY.to_ascii_lowercase()),
"ks-cursor: 42".into(),
"upgrade: websocket".into(),
"connection: upgrade".into(),
] {
assert!(headers.contains(&wanted), "{wanted} in {head}");
}
for unwanted in ["cookie", "x-junk", "x-forwarded-for", "sec-websocket"] {
assert!(
!headers.iter().any(|header| header.starts_with(unwanted)),
"{unwanted} passed on: {head}"
);
}
// nginx drops an empty header. an empty cursor sends none.
let reply = upgrade(port, "GET", "", &websocket(KEY));
assert!(reply.starts_with("HTTP/1.1 101 "), "{reply}");
let head = received.recv_timeout(Duration::from_secs(5)).unwrap();
assert!(!head.to_ascii_lowercase().contains("ks-cursor"), "{head}");
let short_key = websocket("dGhlIHNhbXBsZSBub25jZQ=");
let plain = "Connection: close\r\n".to_string();
for (method, query, headers) in [
("GET", "?cursor=abc", websocket(KEY)),
("GET", "?cursor=-1", websocket(KEY)),
("GET", "?cursor=12345678901234567890", websocket(KEY)),
("GET", "", short_key),
("GET", "", plain),
("POST", "", websocket(KEY)),
] {
let reply = upgrade(port, method, query, &headers);
assert!(
reply.starts_with("HTTP/1.1 400 "),
"{method} {query} {headers:?} must be refused: {reply}"
);
}
assert!(
received.recv_timeout(Duration::from_millis(200)).is_err(),
"a refused request reached the server"
);
}