main / frogue / tests / support / ap.rs
Raw
//! Test helpers for simulating remote ActivityPub actors.
//!
//! RemoteActor is a minimal ActivityPub server that responds to actor fetches
//! and records inbox deliveries. This allows testing the full signature
//! verification and delivery flow without requiring an actual remote server.
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::thread;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64;
use chrono::Utc;
use frogue::activitypub::keys;
use serde_json::json;
use sha2::{Digest, Sha256};
/// A recorded inbox delivery.
#[derive(Debug, Clone)]
pub struct Delivery {
pub signature: String,
pub date: String,
pub digest: String,
pub body: String,
}
/// A recorded GET request.
#[derive(Debug, Clone)]
pub struct GetRequest {
pub path: String,
pub has_signature: bool,
}
/// A minimal remote ActivityPub actor for testing.
pub struct RemoteActor {
pub private_key_pem: String,
pub public_key_pem: String,
pub actor_url: String,
pub inbox_url: String,
pub key_id: String,
pub deliveries: Arc<Mutex<Vec<Delivery>>>,
pub notes: Arc<Mutex<std::collections::HashMap<String, String>>>,
pub get_requests: Arc<Mutex<Vec<GetRequest>>>,
pub gone: bool,
}
/// Multiple remote actors sharing a single inbox listener.
pub struct SharedInboxActors {
pub actors: Vec<RemoteActor>,
pub shared_inbox_url: String,
deliveries: Arc<Mutex<Vec<Delivery>>>,
}
impl SharedInboxActors {
/// Create multiple actors that share a single inbox listener.
/// Each actor has its own keypair and actor document, but they all
/// deliver to the same shared inbox URL.
pub fn serve(port: u16, usernames: &[&str]) -> Self {
let listener =
TcpListener::bind(("127.0.0.1", port)).expect("failed to bind shared inbox listener");
let deliveries = Arc::new(Mutex::new(Vec::new()));
let shared_inbox_url = format!("http://127.0.0.1:{port}/shared-inbox");
// Build actor documents for each username
let mut actors = Vec::new();
let mut actor_docs = Vec::new();
for username in usernames {
let (private_key_pem, public_key_pem) =
keys::generate().expect("key generation failed");
let actor_url = format!("http://127.0.0.1:{port}/actor/{username}");
let key_id = format!("{actor_url}#main-key");
let actor_json = json!({
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1"
],
"id": actor_url,
"type": "Person",
"preferredUsername": username,
"inbox": shared_inbox_url,
"endpoints": {
"sharedInbox": shared_inbox_url
},
"publicKey": {
"id": key_id,
"owner": actor_url,
"publicKeyPem": public_key_pem
}
});
actor_docs.push((
actor_url.clone(),
serde_json::to_string(&actor_json).unwrap(),
));
actors.push(RemoteActor {
private_key_pem,
public_key_pem,
actor_url,
inbox_url: shared_inbox_url.clone(),
key_id,
deliveries: Arc::clone(&deliveries),
notes: Arc::new(Mutex::new(std::collections::HashMap::new())),
get_requests: Arc::new(Mutex::new(Vec::new())),
gone: false,
});
}
let deliveries_clone = Arc::clone(&deliveries);
thread::spawn(move || {
for stream in listener.incoming() {
let stream = match stream {
Ok(s) => s,
Err(_) => continue,
};
let actor_docs = actor_docs.clone();
let deliveries = Arc::clone(&deliveries_clone);
thread::spawn(move || {
handle_shared_inbox_connection(stream, &actor_docs, deliveries);
});
}
});
Self {
actors,
shared_inbox_url,
deliveries,
}
}
/// Get all recorded deliveries.
pub fn recorded_deliveries(&self) -> Vec<Delivery> {
self.deliveries.lock().unwrap().clone()
}
}
/// Handle connections for shared inbox actors.
fn handle_shared_inbox_connection(
mut stream: TcpStream,
actor_docs: &[(String, String)],
deliveries: Arc<Mutex<Vec<Delivery>>>,
) {
let mut buffer = [0; 8192];
let n = match stream.read(&mut buffer) {
Ok(n) => n,
Err(_) => return,
};
let request = String::from_utf8_lossy(&buffer[..n]);
let first_line = request.lines().next().unwrap_or("");
let parts: Vec<&str> = first_line.split_whitespace().collect();
if parts.len() < 2 {
return;
}
let method = parts[0];
let path = parts[1];
let response = if method == "GET" {
// Find the actor document for this path
if let Some((_, actor_json)) = actor_docs.iter().find(|(url, _)| url.ends_with(path)) {
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/activity+json\r\nContent-Length: {}\r\n\r\n{}",
actor_json.len(),
actor_json
)
} else {
"HTTP/1.1 404 Not Found\r\n\r\n".to_string()
}
} else if method == "POST" && (path == "/shared-inbox" || path == "/inbox") {
let (headers, body) = request.split_once("\r\n\r\n").unwrap_or(("", ""));
let mut signature = String::new();
let mut date = String::new();
let mut digest = String::new();
for line in headers.lines().skip(1) {
if let Some((key, value)) = line.split_once(": ") {
match key.to_lowercase().as_str() {
"signature" => signature = value.to_string(),
"date" => date = value.to_string(),
"digest" => digest = value.to_string(),
_ => {}
}
}
}
deliveries.lock().unwrap().push(Delivery {
signature,
date,
digest,
body: body.to_string(),
});
"HTTP/1.1 202 Accepted\r\n\r\n".to_string()
} else {
"HTTP/1.1 404 Not Found\r\n\r\n".to_string()
};
let _ = stream.write_all(response.as_bytes());
}
impl RemoteActor {
/// Create a new remote actor and start listening on the given port.
pub fn serve(port: u16) -> Self {
let (private_key_pem, public_key_pem) = keys::generate().expect("key generation failed");
let actor_url = format!("http://127.0.0.1:{port}/actor");
let inbox_url = format!("http://127.0.0.1:{port}/inbox");
let key_id = format!("{actor_url}#main-key");
let listener =
TcpListener::bind(("127.0.0.1", port)).expect("failed to bind remote actor listener");
let deliveries = Arc::new(Mutex::new(Vec::new()));
let deliveries_clone = Arc::clone(&deliveries);
let notes = Arc::new(Mutex::new(std::collections::HashMap::new()));
let notes_clone = Arc::clone(¬es);
let get_requests = Arc::new(Mutex::new(Vec::new()));
let get_requests_clone = Arc::clone(&get_requests);
// Build the actor JSON for responses
let actor_json = json!({
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1"
],
"id": actor_url,
"type": "Person",
"preferredUsername": "remote",
"name": "Remote Test User",
"inbox": inbox_url,
"publicKey": {
"id": key_id,
"owner": actor_url,
"publicKeyPem": public_key_pem
}
});
let actor_json_str = serde_json::to_string(&actor_json).unwrap();
thread::spawn(move || {
for stream in listener.incoming() {
let stream = match stream {
Ok(s) => s,
Err(_) => continue,
};
let actor_json = actor_json_str.clone();
let deliveries = Arc::clone(&deliveries_clone);
let notes = Arc::clone(¬es_clone);
let get_requests = Arc::clone(&get_requests_clone);
thread::spawn(move || {
handle_connection(stream, &actor_json, deliveries, notes, get_requests, false);
});
}
});
Self {
private_key_pem,
public_key_pem,
actor_url,
inbox_url,
key_id,
deliveries,
notes,
get_requests,
gone: false,
}
}
/// Create a remote actor whose actor fetch returns 410 Gone.
pub fn serve_gone(port: u16) -> Self {
let (private_key_pem, public_key_pem) = keys::generate().expect("key generation failed");
let actor_url = format!("http://127.0.0.1:{port}/actor");
let inbox_url = format!("http://127.0.0.1:{port}/inbox");
let key_id = format!("{actor_url}#main-key");
let listener =
TcpListener::bind(("127.0.0.1", port)).expect("failed to bind remote actor listener");
let deliveries = Arc::new(Mutex::new(Vec::new()));
let deliveries_clone = Arc::clone(&deliveries);
let notes = Arc::new(Mutex::new(std::collections::HashMap::new()));
let notes_clone = Arc::clone(¬es);
let get_requests = Arc::new(Mutex::new(Vec::new()));
let get_requests_clone = Arc::clone(&get_requests);
let actor_json = json!({
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1"
],
"id": actor_url,
"type": "Person",
"preferredUsername": "remote",
"name": "Remote Test User",
"inbox": inbox_url,
"publicKey": {
"id": key_id,
"owner": actor_url,
"publicKeyPem": public_key_pem
}
});
let actor_json_str = serde_json::to_string(&actor_json).unwrap();
let deliveries_for_thread = Arc::clone(&deliveries_clone);
let notes_for_thread = Arc::clone(¬es_clone);
let get_requests_for_thread = Arc::clone(&get_requests_clone);
thread::spawn(move || {
for stream in listener.incoming() {
let stream = match stream {
Ok(s) => s,
Err(_) => continue,
};
let actor_json = actor_json_str.clone();
let deliveries = Arc::clone(&deliveries_for_thread);
let notes = Arc::clone(¬es_for_thread);
let get_requests = Arc::clone(&get_requests_for_thread);
thread::spawn(move || {
handle_connection(stream, &actor_json, deliveries, notes, get_requests, true);
});
}
});
Self {
private_key_pem,
public_key_pem,
actor_url,
inbox_url,
key_id,
deliveries: deliveries_clone,
notes: notes_clone,
get_requests: get_requests_clone,
gone: true,
}
}
/// Create a remote actor that can serve notes at URLs.
/// Notes are added later via serve_note_at_url.
pub fn serve_with_note(port: u16) -> Self {
Self::serve(port)
}
/// Create a remote actor with a shared inbox endpoint.
pub fn serve_with_shared_inbox(port: u16, shared_inbox_url: Option<String>) -> Self {
let (private_key_pem, public_key_pem) = keys::generate().expect("key generation failed");
let actor_url = format!("http://127.0.0.1:{port}/actor");
let inbox_url =
shared_inbox_url.unwrap_or_else(|| format!("http://127.0.0.1:{port}/inbox"));
let key_id = format!("{actor_url}#main-key");
let listener =
TcpListener::bind(("127.0.0.1", port)).expect("failed to bind remote actor listener");
let deliveries = Arc::new(Mutex::new(Vec::new()));
let deliveries_clone = Arc::clone(&deliveries);
let notes = Arc::new(Mutex::new(std::collections::HashMap::new()));
let notes_clone = Arc::clone(¬es);
let get_requests = Arc::new(Mutex::new(Vec::new()));
let get_requests_clone = Arc::clone(&get_requests);
let actor_json = serde_json::json!({
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1"
],
"id": actor_url,
"type": "Person",
"preferredUsername": "remote",
"name": "Remote Test User",
"inbox": inbox_url,
"publicKey": {
"id": key_id,
"owner": actor_url,
"publicKeyPem": public_key_pem
}
});
let actor_json_str = serde_json::to_string(&actor_json).unwrap();
let deliveries_for_thread = Arc::clone(&deliveries_clone);
let notes_for_thread = Arc::clone(¬es_clone);
let get_requests_for_thread = Arc::clone(&get_requests_clone);
thread::spawn(move || {
for stream in listener.incoming() {
let stream = match stream {
Ok(s) => s,
Err(_) => continue,
};
let actor_json = actor_json_str.clone();
let deliveries = Arc::clone(&deliveries_for_thread);
let notes = Arc::clone(¬es_for_thread);
let get_requests = Arc::clone(&get_requests_for_thread);
thread::spawn(move || {
handle_connection(stream, &actor_json, deliveries, notes, get_requests, false);
});
}
});
Self {
private_key_pem,
public_key_pem,
actor_url,
inbox_url,
key_id,
deliveries: deliveries_clone,
notes: notes_clone,
get_requests: get_requests_clone,
gone: false,
}
}
/// Create a remote actor that fails the first N delivery attempts.
pub fn serve_with_failures(port: u16, fail_count: usize) -> Self {
let (private_key_pem, public_key_pem) = keys::generate().expect("key generation failed");
let actor_url = format!("http://127.0.0.1:{port}/actor");
let inbox_url = format!("http://127.0.0.1:{port}/inbox");
let key_id = format!("{actor_url}#main-key");
let listener =
TcpListener::bind(("127.0.0.1", port)).expect("failed to bind remote actor listener");
let deliveries = Arc::new(Mutex::new(Vec::new()));
let deliveries_clone = Arc::clone(&deliveries);
let notes = Arc::new(Mutex::new(std::collections::HashMap::new()));
let notes_clone = Arc::clone(¬es);
let get_requests = Arc::new(Mutex::new(Vec::new()));
let get_requests_clone = Arc::clone(&get_requests);
let fail_counter = Arc::new(Mutex::new(0));
let fail_counter_clone = Arc::clone(&fail_counter);
let actor_json = serde_json::json!({
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1"
],
"id": actor_url,
"type": "Person",
"preferredUsername": "remote",
"name": "Remote Test User",
"inbox": inbox_url,
"publicKey": {
"id": key_id,
"owner": actor_url,
"publicKeyPem": public_key_pem
}
});
let actor_json_str = serde_json::to_string(&actor_json).unwrap();
let deliveries_for_thread = Arc::clone(&deliveries_clone);
let fail_counter_for_thread = Arc::clone(&fail_counter_clone);
thread::spawn(move || {
for stream in listener.incoming() {
let stream = match stream {
Ok(s) => s,
Err(_) => continue,
};
let actor_json = actor_json_str.clone();
let deliveries = Arc::clone(&deliveries_for_thread);
let fail_counter = Arc::clone(&fail_counter_for_thread);
thread::spawn(move || {
handle_connection_with_failures(
stream,
&actor_json,
deliveries,
fail_counter,
fail_count,
);
});
}
});
Self {
private_key_pem,
public_key_pem,
actor_url,
inbox_url,
key_id,
deliveries: deliveries_clone,
notes: notes_clone,
get_requests: get_requests_clone,
gone: false,
}
}
/// Get all recorded deliveries.
pub fn recorded_deliveries(&self) -> Vec<Delivery> {
self.deliveries.lock().unwrap().clone()
}
/// Drain all recorded deliveries, returning them and clearing the queue.
pub fn drain_deliveries(&self) -> Vec<Delivery> {
let mut guard = self.deliveries.lock().unwrap();
std::mem::take(&mut *guard)
}
/// Get all recorded GET requests.
pub fn recorded_get_requests(&self) -> Vec<GetRequest> {
self.get_requests.lock().unwrap().clone()
}
/// Serve a note at a specific URL path.
pub fn serve_note_at_url(&self, path: &str, note_json: serde_json::Value) {
let note_str = serde_json::to_string(¬e_json).unwrap();
self.notes
.lock()
.unwrap()
.insert(path.to_string(), note_str);
}
/// Wait until at least one delivery has been recorded, or timeout.
pub fn wait_for_delivery(&self, timeout_secs: u64) -> Option<Delivery> {
use std::time::{Duration, Instant};
let deadline = Instant::now() + Duration::from_secs(timeout_secs);
while Instant::now() < deadline {
if let Some(delivery) = self.deliveries.lock().unwrap().first().cloned() {
return Some(delivery);
}
thread::sleep(Duration::from_millis(50));
}
None
}
/// Send a Follow activity to the target actor's inbox.
pub fn follow(&self, target_actor_url: String) {
use std::io::Write;
let target_host = target_actor_url
.strip_prefix("http://")
.and_then(|s| s.split('/').next())
.unwrap_or("127.0.0.1");
let inbox_path = format!(
"{}/inbox",
target_actor_url
.strip_prefix("http://")
.unwrap_or(&target_actor_url)
.strip_prefix(&target_host)
.unwrap_or("")
);
let follow_json = serde_json::json!({
"@context": "https://www.w3.org/ns/activitystreams",
"id": format!("{}/follow/{}", self.actor_url, chrono::Utc::now().timestamp()),
"type": "Follow",
"actor": self.actor_url,
"object": target_actor_url
});
let body = serde_json::to_string(&follow_json).unwrap();
let (signature, date, digest) = signed_post(
&self.private_key_pem,
&self.key_id,
target_host,
&inbox_path,
&body,
);
let request = format!(
"POST {} HTTP/1.1\r\n\
Host: {}\r\n\
Date: {}\r\n\
Digest: {}\r\n\
Signature: {}\r\n\
Content-Type: application/activity+json\r\n\
Content-Length: {}\r\n\
\r\n\
{}",
inbox_path,
target_host,
date,
digest,
signature,
body.len(),
body
);
if let Ok(mut stream) = TcpStream::connect(target_host) {
let _ = stream.write_all(request.as_bytes());
}
}
/// Send a generic activity to any inbox URL.
pub fn send_activity(&self, inbox_url: &str, activity: serde_json::Value) {
use std::io::Write;
// Parse host and path from inbox_url
let url_without_scheme = inbox_url.strip_prefix("http://").unwrap_or(inbox_url);
let target_host = url_without_scheme.split('/').next().unwrap_or("127.0.0.1");
let target_path = format!(
"/{}",
url_without_scheme
.split('/')
.skip(1)
.collect::<Vec<_>>()
.join("/")
);
let body = serde_json::to_string(&activity).unwrap();
let (signature, date, digest) = signed_post(
&self.private_key_pem,
&self.key_id,
target_host,
&target_path,
&body,
);
let request = format!(
"POST {} HTTP/1.1\r\n\
Host: {}\r\n\
Date: {}\r\n\
Digest: {}\r\n\
Signature: {}\r\n\
Content-Type: application/activity+json\r\n\
Content-Length: {}\r\n\
\r\n\
{}",
target_path,
target_host,
date,
digest,
signature,
body.len(),
body
);
let mut stream = match TcpStream::connect(target_host) {
Ok(s) => s,
Err(e) => {
eprintln!("Failed to connect to {}: {}", target_host, e);
return;
}
};
if let Err(e) = stream.write_all(request.as_bytes()) {
eprintln!("Failed to write request: {}", e);
}
}
}
impl Drop for RemoteActor {
fn drop(&mut self) {
// The listener will be closed when dropped, ending the thread.
}
}
/// Build signed request headers for sending an activity to a target.
pub fn signed_post(
private_key_pem: &str,
key_id: &str,
target_host: &str,
target_path: &str,
body: &str,
) -> (String, String, String) {
let date = Utc::now().format("%a, %d %b %Y %H:%M:%S GMT").to_string();
let digest = format!("SHA-256={}", BASE64.encode(Sha256::digest(body.as_bytes())));
// Build the signing string
let signing_string = format!(
"(request-target): post {}\nhost: {}\ndate: {}\ndigest: {}\ncontent-type: application/activity+json",
target_path, target_host, date, digest
);
let signature_bytes =
keys::sign(private_key_pem, signing_string.as_bytes()).expect("signing failed");
let signature = format!(
"keyId=\"{}\",headers=\"(request-target) host date digest content-type\",algorithm=\"rsa-sha256\",signature=\"{}\"",
key_id,
BASE64.encode(signature_bytes)
);
(signature, date, digest)
}
/// Build RFC 9421 HTTP Message Signature headers for an outbound POST.
///
/// Returns (Signature, Signature-Input, Date, Content-Digest) headers.
pub fn signed_post_rfc9421(
private_key_pem: &str,
key_id: &str,
target_host: &str,
target_path: &str,
body: &str,
) -> (String, String, String, String) {
let date = Utc::now().format("%a, %d %b %Y %H:%M:%S GMT").to_string();
let created = Utc::now().timestamp();
let body_bytes = body.as_bytes();
let content_digest = format!("sha-256=:{}:", BASE64.encode(Sha256::digest(body_bytes)));
// Build the signature base per RFC 9421
let target_uri = format!("http://{}{}", target_host, target_path);
let signature_base = format!(
"\"@method\": POST\n\
\"@target-uri\": {target_uri}\n\
\"content-digest\": {content_digest}\n\
\"@signature-params\": (\"@method\" \"@target-uri\" \"content-digest\");created={created};keyid=\"{key_id}\";alg=\"rsa-v1_5-sha256\""
);
let signature_bytes =
keys::sign(private_key_pem, signature_base.as_bytes()).expect("signing failed");
let signature_b64 = BASE64.encode(signature_bytes);
let signature_input = format!(
"sig1=(\"@method\" \"@target-uri\" \"content-digest\");created={created};keyid=\"{key_id}\";alg=\"rsa-v1_5-sha256\""
);
let signature = format!("sig1=:{signature_b64}:");
(signature, signature_input, date, content_digest)
}
/// Handle a single connection to the remote actor server.
fn handle_connection(
mut stream: TcpStream,
actor_json: &str,
deliveries: Arc<Mutex<Vec<Delivery>>>,
notes: Arc<Mutex<std::collections::HashMap<String, String>>>,
get_requests: Arc<Mutex<Vec<GetRequest>>>,
gone: bool,
) {
let mut buffer = [0; 8192];
let n = match stream.read(&mut buffer) {
Ok(n) => n,
Err(_) => return,
};
let request = String::from_utf8_lossy(&buffer[..n]);
// Parse the request to determine what to do
let first_line = request.lines().next().unwrap_or("");
let parts: Vec<&str> = first_line.split_whitespace().collect();
if parts.len() < 2 {
return;
}
let method = parts[0];
let path = parts[1];
// Record GET requests
if method == "GET" {
let has_signature = request.to_lowercase().contains("signature:");
get_requests.lock().unwrap().push(GetRequest {
path: path.to_string(),
has_signature,
});
}
let response = if method == "GET" && path == "/actor" {
// If the actor is "gone", return 410 Gone
if gone {
"HTTP/1.1 410 Gone\r\n\r\n".to_string()
} else {
// Respond with the actor document
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/activity+json\r\nContent-Length: {}\r\n\r\n{}",
actor_json.len(),
actor_json
)
}
} else if method == "GET" && path.starts_with("/.well-known/webfinger") {
// Extract the resource parameter from the query string
if let Some(query) = path.split('?').nth(1) {
let params: std::collections::HashMap<&str, &str> = query
.split('&')
.filter_map(|pair| pair.split_once('='))
.collect();
if let Some(resource) = params.get("resource") {
// Parse acct:user@host format
if let Some(acct) = resource.strip_prefix("acct:") {
if let Some((username, _host)) = acct.split_once('@') {
if username == "remote" {
// Return a JRD pointing to the actor URL
let jrd = serde_json::json!({
"subject": resource,
"links": [{
"rel": "self",
"type": "application/activity+json",
"href": "http://127.0.0.1:PORT/actor"
}]
});
// Replace PORT placeholder with actual port
let jrd_str = jrd
.to_string()
.replace("PORT", &stream.local_addr().unwrap().port().to_string());
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/jrd+json\r\nContent-Length: {}\r\n\r\n{}",
jrd_str.len(),
jrd_str
)
} else {
"HTTP/1.1 404 Not Found\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} else if method == "GET" {
// Check if this is a request for a note
let notes_lock = notes.lock().unwrap();
if let Some(note_json) = notes_lock.get(path) {
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/activity+json\r\nContent-Length: {}\r\n\r\n{}",
note_json.len(),
note_json
)
} else {
"HTTP/1.1 404 Not Found\r\n\r\n".to_string()
}
} else if method == "POST" && path.ends_with("/inbox") {
// Extract headers and body
let (headers, body) = request.split_once("\r\n\r\n").unwrap_or(("", ""));
let mut signature = String::new();
let mut date = String::new();
let mut digest = String::new();
for line in headers.lines().skip(1) {
if let Some((key, value)) = line.split_once(": ") {
match key.to_lowercase().as_str() {
"signature" => signature = value.to_string(),
"date" => date = value.to_string(),
"digest" => digest = value.to_string(),
_ => {}
}
}
}
deliveries.lock().unwrap().push(Delivery {
signature,
date,
digest,
body: body.to_string(),
});
"HTTP/1.1 202 Accepted\r\n\r\n".to_string()
} else {
"HTTP/1.1 404 Not Found\r\n\r\n".to_string()
};
let _ = stream.write_all(response.as_bytes());
}
/// Handle a connection with configurable failures for the first N POST
/// requests.
fn handle_connection_with_failures(
mut stream: TcpStream,
actor_json: &str,
deliveries: Arc<Mutex<Vec<Delivery>>>,
fail_counter: Arc<Mutex<usize>>,
fail_count: usize,
) {
let mut buffer = [0; 8192];
let n = match stream.read(&mut buffer) {
Ok(n) => n,
Err(_) => return,
};
let request = String::from_utf8_lossy(&buffer[..n]);
// Parse the request to determine what to do
let first_line = request.lines().next().unwrap_or("");
let parts: Vec<&str> = first_line.split_whitespace().collect();
if parts.len() < 2 {
return;
}
let method = parts[0];
let path = parts[1];
let response = if method == "GET" && path == "/actor" {
// Respond with the actor document
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/activity+json\r\nContent-Length: {}\r\n\r\n{}",
actor_json.len(),
actor_json
)
} else if method == "GET" && path.starts_with("/.well-known/webfinger") {
// Parse query parameters
let query = path.split('?').nth(1).unwrap_or("");
let params: std::collections::HashMap<&str, &str> = query
.split('&')
.filter_map(|pair| pair.split_once('='))
.collect();
if let Some(resource) = params.get("resource") {
// Parse acct:username@host format
if let Some(acct) = resource.strip_prefix("acct:") {
if let Some((username, _host)) = acct.split_once('@') {
if username == "remote" {
// Return a JRD pointing to the actor URL
let jrd = serde_json::json!({
"subject": resource,
"links": [{
"rel": "self",
"type": "application/activity+json",
"href": "http://127.0.0.1:PORT/actor"
}]
});
// Replace PORT placeholder with actual port
let jrd_str = jrd
.to_string()
.replace("PORT", &stream.local_addr().unwrap().port().to_string());
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/jrd+json\r\nContent-Length: {}\r\n\r\n{}",
jrd_str.len(),
jrd_str
)
} else {
"HTTP/1.1 404 Not Found\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 400 Bad Request\r\n\r\n".to_string()
}
} else if method == "POST" && path == "/inbox" {
// Check if we should fail this request
let should_fail = {
let mut counter = fail_counter.lock().unwrap();
if *counter < fail_count {
*counter += 1;
true
} else {
false
}
};
if should_fail {
// Return 500 error
"HTTP/1.1 500 Internal Server Error\r\n\r\n".to_string()
} else {
// Extract headers and body
let (headers, body) = request.split_once("\r\n\r\n").unwrap_or(("", ""));
let mut signature = String::new();
let mut date = String::new();
let mut digest = String::new();
for line in headers.lines().skip(1) {
if let Some((key, value)) = line.split_once(": ") {
match key.to_lowercase().as_str() {
"signature" => signature = value.to_string(),
"date" => date = value.to_string(),
"digest" => digest = value.to_string(),
_ => {}
}
}
}
deliveries.lock().unwrap().push(Delivery {
signature,
date,
digest,
body: body.to_string(),
});
"HTTP/1.1 202 Accepted\r\n\r\n".to_string()
}
} else {
"HTTP/1.1 404 Not Found\r\n\r\n".to_string()
};
let _ = stream.write_all(response.as_bytes());
}