reactor
/
ptth
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

wip: debug proxy now owns a filter which can drop or modify request bodies

main
_ 2021-03-06 22:58:23 +00:00
parent a980d151fc
commit 9648a9853c
5 changed files with 224 additions and 184 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Cargo.lock generated
View File

@ -165,9 +165,9 @@ checksum = "e91831deabf0d6d7ec49552e489aed63b7456a7a3c46cff62adad428110b0af0"
[[package]] [[package]]
name = "async-trait" name = "async-trait"
version = "0.1.42" version = "0.1.45"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8d3a45e77e34375a7923b1e8febb049bb011f064714a8e17a1a616fef01da13d" checksum = "d3340571769500ddef1e94b45055fabed6b08a881269b7570c830b8f32ef84ef"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@ -468,6 +468,7 @@ name = "debug_proxy"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait",
"futures-util", "futures-util",
"http", "http",
"hyper", "hyper",

1
crates/debug_proxy/Cargo.toml
View File

@ -9,6 +9,7 @@ license = "AGPL-3.0"
[dependencies] [dependencies]
anyhow = "1.0.34" anyhow = "1.0.34"
async-trait = "0.1.45"
futures-util = "0.3.8" futures-util = "0.3.8"
http = "0.2.1" http = "0.2.1"
hyper = { version = "0.14.4", features = ["server", "stream"] } hyper = { version = "0.14.4", features = ["server", "stream"] }

98
crates/debug_proxy/src/lib.rs
View File

@ -3,12 +3,14 @@ use std::{
sync::Arc, sync::Arc,
}; };
use async_trait::async_trait;
use futures_util::StreamExt; use futures_util::StreamExt;
use hyper::{ use hyper::{
Body, Body,
Request, Request,
Response, Response,
Server, Server,
body::Bytes,
service::{ service::{
make_service_fn, make_service_fn,
service_fn, service_fn,
@ -25,18 +27,77 @@ use tokio::{
use tokio_stream::wrappers::ReceiverStream; use tokio_stream::wrappers::ReceiverStream;
use ulid::Ulid; use ulid::Ulid;
struct State { #[async_trait]
client: Client, trait ProxyFilter {
upstream_authority: String, async fn request_body (&self, mut body: Body, tx: mpsc::Sender <Result <Bytes, hyper::Error>>) -> anyhow::Result <()>;
} }
async fn handle_all (req: Request <Body>, state: Arc <State>) struct PassthroughFilter {}
#[async_trait]
impl ProxyFilter for PassthroughFilter {
async fn request_body (&self, mut body: Body, tx: mpsc::Sender <Result <Bytes, hyper::Error>>) -> anyhow::Result <()> {
let mut bytes_transferred = 0;
loop {
let item = body.next ().await;
if let Some (item) = item {
if let Ok (item) = &item {
bytes_transferred += item.len ();
}
tx.send (item).await?;
}
else {
// Finished
break;
}
}
Ok (())
}
}
struct RequestBodyDropFilter {}
#[async_trait]
impl ProxyFilter for RequestBodyDropFilter {
async fn request_body (&self, mut body: Body, tx: mpsc::Sender <Result <Bytes, hyper::Error>>) -> anyhow::Result <()> {
let mut bytes_transferred = 0;
loop {
let item = body.next ().await;
if let Some (item) = item {
if let Ok (item) = &item {
bytes_transferred += item.len ();
}
// tx.send (item).await?;
tracing::debug! ("RequestBodyDropFilter dropping chunk");
}
else {
// Finished
break;
}
}
Ok (())
}
}
struct State <PF> {
client: Client,
upstream_authority: String,
proxy_filter: Arc <PF>,
}
async fn handle_all <PF: 'static + ProxyFilter + Sync + Send> (req: Request <Body>, state: Arc <State <PF>>)
-> anyhow::Result <Response <Body>> -> anyhow::Result <Response <Body>>
{ {
let req_id = Ulid::new ().to_string (); let req_id = Ulid::new ().to_string ();
let (head, mut body) = req.into_parts (); let (head, mut body) = req.into_parts ();
tracing::trace! ("{} Got URI {}", req_id, head.uri); tracing::debug! ("{} Got URI {}", req_id, head.uri);
let upstream_authority = state.upstream_authority.clone (); let upstream_authority = state.upstream_authority.clone ();
@ -55,27 +116,10 @@ async fn handle_all (req: Request <Body>, state: Arc <State>)
let (tx, rx) = mpsc::channel (1); let (tx, rx) = mpsc::channel (1);
spawn ({ spawn ({
let req_id = req_id.clone (); let req_id = req_id.clone ();
let proxy_filter = state.proxy_filter.clone ();
async move { async move {
let mut bytes_transferred = 0; proxy_filter.request_body (body, tx).await
loop {
let item = body.next ().await;
if let Some (item) = item {
if let Ok (item) = &item {
bytes_transferred += item.len ();
}
tx.send (item).await?;
}
else {
// Finished
break;
}
}
tracing::trace! ("{} Request body bytes: {}", req_id, bytes_transferred);
Ok::<_, anyhow::Error> (())
} }
}); });
@ -122,9 +166,13 @@ pub async fn run_proxy (
shutdown_oneshot: oneshot::Receiver <()>, shutdown_oneshot: oneshot::Receiver <()>,
) -> anyhow::Result <()> ) -> anyhow::Result <()>
{ {
let filter = PassthroughFilter {};
// let filter = RequestBodyDropFilter {};
let state = Arc::new (State { let state = Arc::new (State {
client: Client::builder ().build ()?, client: Client::builder ().build ()?,
upstream_authority, upstream_authority,
proxy_filter: Arc::new (filter),
}); });
let make_svc = make_service_fn (|_conn| { let make_svc = make_service_fn (|_conn| {

5
crates/ptth_server/src/lib.rs
View File

@ -106,9 +106,8 @@ async fn handle_one_req (
if e.is_request () { if e.is_request () {
warn! ("Error while POSTing response. Client probably hung up."); warn! ("Error while POSTing response. Client probably hung up.");
} }
else {
error! ("Err: {:?}", e); error! ("Err: {:?}", e);
}
}, },
} }

267
src/tests.rs
View File

@ -9,7 +9,6 @@ use std::{
}; };
use tokio::{ use tokio::{
runtime::Runtime,
spawn, spawn,
sync::oneshot, sync::oneshot,
}; };
@ -42,8 +41,9 @@ async fn testing_client_checks (
assert_eq! (resp, "Relay is up\n"); assert_eq! (resp, "Relay is up\n");
let resp = client.get (&format! ("{}/frontend/servers/{}/files/COPYING", relay_url, server_name)) let req = client.get (&format! ("{}/frontend/servers/{}/files/COPYING", relay_url, server_name))
.send ().await.expect ("Couldn't find license").bytes ().await.expect ("Couldn't find license"); .send ();
let resp = tokio::time::timeout (Duration::from_secs (2), req).await.expect ("Request timed out").expect ("Couldn't find license").bytes ().await.expect ("Couldn't find license");
if blake3::hash (&resp) != blake3::Hash::from ([ if blake3::hash (&resp) != blake3::Hash::from ([
0xca, 0x02, 0x92, 0x78, 0xca, 0x02, 0x92, 0x78,
@ -174,172 +174,163 @@ impl TestingServer {
} }
} }
#[test] #[tokio::test]
fn end_to_end () { async fn end_to_end () {
// Prefer this form for tests, since all tests share one process // Prefer this form for tests, since all tests share one process
// and we don't care if another test already installed a subscriber. // and we don't care if another test already installed a subscriber.
//tracing_subscriber::fmt ().try_init ().ok (); //tracing_subscriber::fmt ().try_init ().ok ();
let rt = Runtime::new ().expect ("Can't create runtime for testing");
// Spawn the root task let relay_port = 4000;
rt.block_on (async { // No proxy
let relay_port = 4000; let proxy_port = relay_port;
// No proxy let server_name = "aliens_wildland";
let proxy_port = relay_port;
let server_name = "aliens_wildland";
let testing_config = TestingConfig { let testing_config = TestingConfig {
server_name, server_name,
api_key: "AnacondaHardcoverGrannyUnlatchLankinessMutate", api_key: "AnacondaHardcoverGrannyUnlatchLankinessMutate",
proxy_port, proxy_port,
relay_port, relay_port,
}; };
let testing_relay = TestingRelay::new (&testing_config).await; let testing_relay = TestingRelay::new (&testing_config).await;
let testing_server = TestingServer::new (&testing_config).await; let testing_server = TestingServer::new (&testing_config).await;
wait_for_any_server (&testing_relay.state).await; wait_for_any_server (&testing_relay.state).await;
assert_eq! (testing_relay.state.list_servers ().await, vec! [ assert_eq! (testing_relay.state.list_servers ().await, vec! [
server_name.to_string (), server_name.to_string (),
]); ]);
let client = Client::builder () let client = Client::builder ()
.build ().expect ("Couldn't build HTTP client"); .build ().expect ("Couldn't build HTTP client");
testing_client_checks (&testing_config, &client).await; testing_client_checks (&testing_config, &client).await;
info! ("Shutting down end-to-end test"); info! ("Shutting down end-to-end test");
testing_server.graceful_shutdown ().await; testing_server.graceful_shutdown ().await;
testing_relay.graceful_shutdown ().await; testing_relay.graceful_shutdown ().await;
});
} }
#[test] #[tokio::test]
fn debug_proxy () { async fn debug_proxy () {
tracing_subscriber::fmt ().try_init ().ok (); tracing_subscriber::fmt ()
let rt = Runtime::new ().expect ("Can't create runtime for testing"); .with_env_filter (tracing_subscriber::EnvFilter::from_default_env ())
.try_init ().ok ();
rt.block_on (async { let relay_port = 4002;
let relay_port = 4002; let proxy_port = 11510;
let proxy_port = 11510;
// Start relay // Start relay
let server_name = "aliens_wildland"; let server_name = "aliens_wildland";
let testing_config = TestingConfig { let testing_config = TestingConfig {
server_name, server_name,
api_key: "AnacondaHardcoverGrannyUnlatchLankinessMutate", api_key: "AnacondaHardcoverGrannyUnlatchLankinessMutate",
proxy_port, proxy_port,
relay_port, relay_port,
}; };
let testing_relay = TestingRelay::new (&testing_config).await; let testing_relay = TestingRelay::new (&testing_config).await;
// Start proxy // Start proxy
let (stop_proxy_tx, stop_proxy_rx) = oneshot::channel (); let (stop_proxy_tx, stop_proxy_rx) = oneshot::channel ();
let task_proxy = spawn (async move { let task_proxy = spawn (async move {
debug_proxy::run_proxy (SocketAddr::from (([0, 0, 0, 0], proxy_port)), format! ("127.0.0.1:{}", relay_port), stop_proxy_rx).await debug_proxy::run_proxy (SocketAddr::from (([0, 0, 0, 0], proxy_port)), format! ("127.0.0.1:{}", relay_port), stop_proxy_rx).await
});
// Start server
let testing_server = TestingServer::new (&testing_config).await;
wait_for_any_server (&testing_relay.state).await;
assert_eq! (testing_relay.state.list_servers ().await, vec! [
server_name.to_string (),
]);
let client = Client::builder ()
.build ().expect ("Couldn't build HTTP client");
testing_client_checks (&testing_config, &client).await;
info! ("Shutting down end-to-end test");
testing_server.graceful_shutdown ().await;
stop_proxy_tx.send (()).expect ("Couldn't shut down proxy");
task_proxy.await.expect ("Couldn't join proxy").expect ("Proxy error");
info! ("Proxy stopped");
testing_relay.graceful_shutdown ().await;
}); });
// Start server
let testing_server = TestingServer::new (&testing_config).await;
wait_for_any_server (&testing_relay.state).await;
assert_eq! (testing_relay.state.list_servers ().await, vec! [
server_name.to_string (),
]);
let client = Client::builder ()
.build ().expect ("Couldn't build HTTP client");
testing_client_checks (&testing_config, &client).await;
info! ("Shutting down end-to-end test");
testing_server.graceful_shutdown ().await;
stop_proxy_tx.send (()).expect ("Couldn't shut down proxy");
task_proxy.await.expect ("Couldn't join proxy").expect ("Proxy error");
info! ("Proxy stopped");
testing_relay.graceful_shutdown ().await;
} }
#[test] #[tokio::test]
fn scraper_endpoints () { async fn scraper_endpoints () {
let rt = Runtime::new ().expect ("Can't create runtime for testing"); use ptth_relay::*;
rt.block_on (async { let config_file = config::file::Config {
use ptth_relay::*; iso: config::file::Isomorphic {
enable_scraper_api: true,
dev_mode: Default::default (),
},
port: Some (4001),
servers: vec! [],
scraper_keys: vec! [
key_validity::ScraperKey::new_30_day ("automated testing", b"bogus")
],
};
let config_file = config::file::Config { let config = config::Config::try_from (config_file).expect ("Can't load config");
iso: config::file::Isomorphic {
enable_scraper_api: true, let relay_state = Arc::new (RelayState::try_from (config).expect ("Can't create relay state"));
dev_mode: Default::default (), let relay_state_2 = relay_state.clone ();
let (stop_relay_tx, stop_relay_rx) = oneshot::channel ();
let task_relay = spawn (async move {
run_relay (
relay_state_2,
Arc::new (load_templates (&PathBuf::new ())?),
stop_relay_rx,
None
).await
});
let relay_url = "http://127.0.0.1:4001";
let mut headers = reqwest::header::HeaderMap::new ();
headers.insert ("X-ApiKey", "bogus".try_into ().unwrap ());
let client = Client::builder ()
.default_headers (headers)
.timeout (Duration::from_secs (2))
.build ().expect ("Couldn't build HTTP client");
let mut resp = None;
for _ in 0usize..5 {
let x = client.get (&format! ("{}/scraper/api/test", relay_url))
.send ().await;
match x {
Err (_) => {
// Probably a reqwest error cause the port is in
// use or something. Try again.
},
Ok (x) => {
resp = Some (x);
break;
}, },
port: Some (4001),
servers: vec! [],
scraper_keys: vec! [
key_validity::ScraperKey::new_30_day ("automated testing", b"bogus")
],
}; };
let config = config::Config::try_from (config_file).expect ("Can't load config"); tokio::time::sleep (Duration::from_millis (200)).await;
}
let resp = resp.expect ("Reqwest repeatedly failed to connect to the relay");
let resp = resp.bytes ().await.expect ("Couldn't check if relay is up");
let relay_state = Arc::new (RelayState::try_from (config).expect ("Can't create relay state")); assert_eq! (resp, "You're valid!\n");
let relay_state_2 = relay_state.clone ();
let (stop_relay_tx, stop_relay_rx) = oneshot::channel ();
let task_relay = spawn (async move {
run_relay (
relay_state_2,
Arc::new (load_templates (&PathBuf::new ())?),
stop_relay_rx,
None
).await
});
let relay_url = "http://127.0.0.1:4001"; stop_relay_tx.send (()).expect ("Couldn't shut down relay");
task_relay.await.expect ("Couldn't join relay").expect ("Relay error");
let mut headers = reqwest::header::HeaderMap::new ();
headers.insert ("X-ApiKey", "bogus".try_into ().unwrap ());
let client = Client::builder ()
.default_headers (headers)
.timeout (Duration::from_secs (2))
.build ().expect ("Couldn't build HTTP client");
let mut resp = None;
for _ in 0usize..5 {
let x = client.get (&format! ("{}/scraper/api/test", relay_url))
.send ().await;
match x {
Err (_) => {
// Probably a reqwest error cause the port is in
// use or something. Try again.
},
Ok (x) => {
resp = Some (x);
break;
},
};
tokio::time::sleep (Duration::from_millis (200)).await;
}
let resp = resp.expect ("Reqwest repeatedly failed to connect to the relay");
let resp = resp.bytes ().await.expect ("Couldn't check if relay is up");
assert_eq! (resp, "You're valid!\n");
stop_relay_tx.send (()).expect ("Couldn't shut down relay");
task_relay.await.expect ("Couldn't join relay").expect ("Relay error");
});
} }