diff --git a/apps/yaak-proxy/components/FlowDetail.tsx b/apps/yaak-proxy/components/FlowDetail.tsx index 5064045a6..5cb5e818d 100644 --- a/apps/yaak-proxy/components/FlowDetail.tsx +++ b/apps/yaak-proxy/components/FlowDetail.tsx @@ -69,8 +69,8 @@ export function FlowDetail({ style, sideBySide }: { style?: CSSProperties; sideB function FlowPane({ flow }: { flow: Flow }) { const [pretty, setPretty] = useState(true); const responseBody = useFlowBody(flow, "response", flow.responseTruncated); - // Failed before any response; an error after the headers is a body cut short mid-way. - const noResponse = flow.error != null && flow.status == null; + // Failed or cancelled before any response; after the headers, the body was cut short mid-way. + const noResponse = (flow.error != null || flow.cancelled) && flow.status == null; const mime = mimeEssence(findHeader(flow.responseHeaders, "content-type")); const queryCount = useMemo(() => queryParams(flow.url).length, [flow.url]); const openSave = useSetAtom(saveDialogFlowAtom); @@ -133,11 +133,15 @@ function FlowPane({ flow }: { flow: Flow }) { )} > - + {flow.cancelled && flow.status == null ? ( + Cancelled + ) : ( + + )} {flow.elapsedMs != null && ( <> • @@ -197,6 +201,13 @@ function FlowPane({ flow }: { flow: Flow }) { {flow.error} )} + {flow.cancelled && ( + + {flow.status == null + ? "The app cancelled the request before a response came back." + : "The app cancelled the request before the response finished."} + + )} {noResponse ? ( -
No response
+
+ {flow.cancelled ? "Cancelled" : "No response"} +
- The request never got a response. The Request tab has what the app sent. + {flow.cancelled + ? "The app cancelled the request before a response came back." + : "The request never got a response."}{" "} + The Request tab has what the app sent.
) : ( diff --git a/apps/yaak-proxy/components/FlowList.tsx b/apps/yaak-proxy/components/FlowList.tsx index 2d5b03bb4..9308e7e6d 100644 --- a/apps/yaak-proxy/components/FlowList.tsx +++ b/apps/yaak-proxy/components/FlowList.tsx @@ -315,8 +315,8 @@ const FlowRow = memo(function FlowRow({ }: RowProps) { const { host, path, query } = splitUrl(flow.url); const failed = flow.error != null; - // Failed before any response; an error after the headers is a body cut short mid-way. - const noResponse = failed && flow.status == null; + // Failed or cancelled before any response; after the headers, the body was cut short mid-way. + const noResponse = (failed || flow.cancelled) && flow.status == null; const type = shortContentType(flow.contentType); return ( @@ -342,6 +342,8 @@ const FlowRow = memo(function FlowRow({ ) : failed ? ( + ) : flow.cancelled ? ( + ) : flow.responseTruncated ? ( ) : null} @@ -362,12 +364,18 @@ const FlowRow = memo(function FlowRow({ {path} {query} - + {flow.cancelled && flow.status == null ? ( + -- + ) : ( + + )} {!sideBySide && (failed ? ( {flow.error} + ) : flow.cancelled ? ( + Cancelled ) : ( {type ?? --} diff --git a/apps/yaak-proxy/components/ProxyStatus.tsx b/apps/yaak-proxy/components/ProxyStatus.tsx index fed381640..4c92bd8d1 100644 --- a/apps/yaak-proxy/components/ProxyStatus.tsx +++ b/apps/yaak-proxy/components/ProxyStatus.tsx @@ -21,7 +21,7 @@ import { recordingLabel, systemProxySwitch } from "../lib/systemProxy"; import { useShowToast } from "../lib/toast"; /** - * The header's state pill and its menu: recording, the system proxy and the certificate. The + * The header's state pill and its menu: recording, the system proxy, the certificate and HTTP/2. The * engine runs for as long as the app does, so the pill shows whether flows are being stored. A * certificate that is not trusted steps out of the menu beside the pill until it is. */ @@ -32,6 +32,7 @@ export function ProxyStatus() { const execute = useRpcMutation("execute_action"); const retry = useRpcMutation("take_over"); const setAtLaunch = useSetSystemProxyAtLaunch(); + const setHttpVersion = useSetHttpVersion(); const copyEnvironment = useCopyEnvironment(); const { caStatus, trust, systemProxy, setSystemProxy } = setup; const systemSwitch = useSystemProxySwitch(setup); @@ -168,6 +169,30 @@ export function ProxyStatus() { /> ), }, + { + type: "content", + label: ( + } + title="HTTP/2" + description="Lets browsers send many requests at once. Turn off for apps that misbehave on it. Applies to new connections." + action={ + + + setHttpVersion.mutate({ httpVersion: enabled ? "auto" : "http1Only" }) + } + /> + + } + /> + ), + }, { type: "separator" }, { label: "Copy Environment for a Terminal", @@ -265,6 +290,14 @@ function useSetSystemProxyAtLaunch() { }); } +/** Saves the HTTP versions offered to clients, which the engine applies to new connections */ +function useSetHttpVersion() { + const store = useStore(); + return useRpcMutation("set_http_version", { + onSettled: () => void refreshProxyState(store), + }); +} + export function useCopyEnvironment() { const proxyState = useAtomValue(proxyStateAtom); const caStatus = useRpcQuery("ca_status", {}); diff --git a/apps/yaak-proxy/lib/flowActions.ts b/apps/yaak-proxy/lib/flowActions.ts index 4c91d67fc..61d64848c 100644 --- a/apps/yaak-proxy/lib/flowActions.ts +++ b/apps/yaak-proxy/lib/flowActions.ts @@ -54,7 +54,7 @@ export function useFlowActions(): (flow: Flow) => DropdownItem[] { showToast({ message: `Failed to copy ${label}: ${String(err)}`, color: "danger" }); } }; - const noResponse = flow.error != null && flow.status == null; + const noResponse = (flow.error != null || flow.cancelled) && flow.status == null; return [ { label: flow.savedRequestId == null ? "Save as Yaak Request" : "Save to Yaak Again", diff --git a/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts b/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts index 5f8202d2a..153dd80f9 100644 --- a/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts +++ b/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts @@ -41,6 +41,11 @@ elapsedHeadersMs: number | null, * Milliseconds from `started_at` to the end of the response. */ elapsedMs: number | null, error: string | null, +/** + * The client gave up before the response ended, by resetting its HTTP/2 stream or closing + * the connection. Not an error. + */ +cancelled: boolean, /** * The id of the Yaak request this flow was last saved as. */ diff --git a/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts b/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts index 07e1a4a7f..68a3b5fa4 100644 --- a/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts +++ b/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts @@ -163,6 +163,10 @@ recording: boolean, * The system proxy turns on as the engine starts. */ systemProxyAtLaunch: boolean, +/** + * The HTTP versions offered to clients. + */ +httpVersion: HttpVersion, /** * Why the engine last failed to start, until it starts. */ @@ -175,6 +179,12 @@ otherOwner: OtherOwner | null, }; export type GlobalAction = "recording_toggle" | "flows_clear"; +/** + * The HTTP versions offered to clients in intercepted HTTPS. Upstream, the engine speaks + * whichever the server picks either way. + */ +export type HttpVersion = "auto" | "http1Only"; + export type ListActionsRequest = Record; export type ListActionsResponse = { actions: Array<[ActionInvocation, ActionMetadata]>, }; @@ -222,7 +232,7 @@ export type ProxyStatePayload = { state: ProxyState, }; export type RpcEventSchema = { flows_cleared: FlowsClearedPayload, model_write: ModelPayload, proxy_state_changed: ProxyStatePayload, }; -export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], list_actions: [ListActionsRequest, ListActionsResponse], list_flows: [FlowFilter, ListFlowsResponse], set_recording: [SetRecordingRequest, boolean], set_system_proxy_at_launch: [SetSystemProxyAtLaunchRequest, boolean], app_icon: [AppIconRequest, string | null], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], clear_flows: [ClearFlowsRequest, number], export_har: [ExportHarRequest, number], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_workspaces: [ListWorkspacesRequest, ListWorkspacesResponse], preview_save: [PreviewSaveRequest, SavePreview], save_flow: [SaveFlowToYaakRequest, SavedFlow], system_proxy_set: [SystemProxySetRequest, SystemProxyStatus], system_proxy_status: [SystemProxyStatusRequest, SystemProxyStatus], take_over: [TakeOverRequest, boolean], }; +export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], list_actions: [ListActionsRequest, ListActionsResponse], list_flows: [FlowFilter, ListFlowsResponse], set_recording: [SetRecordingRequest, boolean], set_system_proxy_at_launch: [SetSystemProxyAtLaunchRequest, boolean], set_http_version: [SetHttpVersionRequest, HttpVersion], app_icon: [AppIconRequest, string | null], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], clear_flows: [ClearFlowsRequest, number], export_har: [ExportHarRequest, number], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_workspaces: [ListWorkspacesRequest, ListWorkspacesResponse], preview_save: [PreviewSaveRequest, SavePreview], save_flow: [SaveFlowToYaakRequest, SavedFlow], system_proxy_set: [SystemProxySetRequest, SystemProxyStatus], system_proxy_status: [SystemProxyStatusRequest, SystemProxyStatus], take_over: [TakeOverRequest, boolean], }; export type SaveFlowToYaakRequest = { flowId: string, workspaceId: string, folderId?: string, /** @@ -272,6 +282,8 @@ export type ServiceProxy = { */ name: string, http: ProxySetting, https: ProxySetting, }; +export type SetHttpVersionRequest = { httpVersion: HttpVersion, }; + export type SetRecordingRequest = { enabled: boolean, }; export type SetSystemProxyAtLaunchRequest = { enabled: boolean, }; diff --git a/crates-proxy/yaak-proxy-lib/migrations/00000005_flow_cancelled.sql b/crates-proxy/yaak-proxy-lib/migrations/00000005_flow_cancelled.sql new file mode 100644 index 000000000..5f9e799a6 --- /dev/null +++ b/crates-proxy/yaak-proxy-lib/migrations/00000005_flow_cancelled.sql @@ -0,0 +1,2 @@ +-- The client gave up on the flow, by resetting its HTTP/2 stream or closing the connection. +ALTER TABLE flows ADD COLUMN cancelled BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/crates-proxy/yaak-proxy-lib/src/bodies.rs b/crates-proxy/yaak-proxy-lib/src/bodies.rs index d0bd60731..ed3f07abf 100644 --- a/crates-proxy/yaak-proxy-lib/src/bodies.rs +++ b/crates-proxy/yaak-proxy-lib/src/bodies.rs @@ -37,8 +37,10 @@ pub async fn read_body( ) -> Result { let raw = read_raw_body(blobs, &flow.id, body)?; let stored_size = raw.len(); - // A response that failed after its headers ended short on the wire. - let cut = body == FlowBody::Response && flow.status.is_some() && flow.error.is_some(); + // A response that failed or was cancelled after its headers ended short on the wire. + let cut = body == FlowBody::Response + && flow.status.is_some() + && (flow.error.is_some() || flow.cancelled); let wire_truncated = flow.truncated(body) || cut; let encoding = ContentEncoding::from_header(find_header(flow.headers(body), "content-encoding")); @@ -90,7 +92,39 @@ mod tests { let request = read_body(&blobs, &cut, FlowBody::Request, 1024).await.unwrap(); assert!(!request.truncated); // With no headers there was never a response body to cut. - let refused = Flow { status: None, ..cut }; + let refused = Flow { status: None, ..cut.clone() }; assert!(!read_body(&blobs, &refused, FlowBody::Response, 1024).await.unwrap().truncated); + + // The client cancelling partway through cuts it short the same way. + let cancelled = Flow { error: None, cancelled: true, ..cut }; + let body = read_body(&blobs, &cancelled, FlowBody::Response, 1024).await.unwrap(); + assert_eq!((body.data.as_slice(), body.truncated), (&b"{\"rows\": ["[..], true)); + } + + #[tokio::test] + async fn cancelled_gzip_response_keeps_what_decoded() { + use std::io::Write; + let (_db, blobs, _rx) = yaak_models::init_in_memory().unwrap(); + let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + gzip.write_all(&[b'z'; 4096]).unwrap(); + let gzip = gzip.finish().unwrap(); + let mut writer = + BodyChunkWriter::new(blobs.clone(), Flow::body_id("fl_gz", FlowBody::Response)); + writer.write(&gzip[..gzip.len() / 2]).unwrap(); + writer.finish().unwrap(); + let flow = Flow { + id: "fl_gz".into(), + status: Some(200), + cancelled: true, + response_headers: vec![crate::models::ProxyHeader { + name: "content-encoding".into(), + value: "gzip".into(), + }], + ..Default::default() + }; + + let body = read_body(&blobs, &flow, FlowBody::Response, 8192).await.unwrap(); + assert!(body.truncated); + assert_eq!(body.decode_error, None); } } diff --git a/crates-proxy/yaak-proxy-lib/src/db.rs b/crates-proxy/yaak-proxy-lib/src/db.rs index 2ae3b8423..dcbb88a59 100644 --- a/crates-proxy/yaak-proxy-lib/src/db.rs +++ b/crates-proxy/yaak-proxy-lib/src/db.rs @@ -140,6 +140,7 @@ mod tests { ALTER TABLE flows DROP COLUMN client_name; ALTER TABLE flows DROP COLUMN client_bundle_id; ALTER TABLE flows DROP COLUMN client_path; + ALTER TABLE flows DROP COLUMN cancelled; INSERT INTO flows (id, url, method) VALUES ('fl_old', 'https://example.com', 'GET');", ) .unwrap(); @@ -150,6 +151,7 @@ mod tests { assert_eq!(flows.len(), 1); assert_eq!((flows[0].id.as_str(), flows[0].saved_request_id.as_deref()), ("fl_old", None)); assert_eq!((flows[0].client_pid, flows[0].client_name.as_deref()), (None, None)); + assert!(!flows[0].cancelled); } #[test] @@ -194,6 +196,7 @@ mod tests { "client_name", "client_bundle_id", "client_path", + "cancelled", ] { assert!(flow_columns.contains(&column.to_string()), "{column} in {flow_columns:?}"); } diff --git a/crates-proxy/yaak-proxy-lib/src/lib.rs b/crates-proxy/yaak-proxy-lib/src/lib.rs index d57cc3dac..45897614f 100644 --- a/crates-proxy/yaak-proxy-lib/src/lib.rs +++ b/crates-proxy/yaak-proxy-lib/src/lib.rs @@ -23,7 +23,7 @@ use crate::db::ProxyQueryManager; use crate::macos::{SystemProxyBackend, SystemProxyStatus}; use crate::models::{Flow, FlowBody, FlowIden, ModelPayload, ProxyHeader, find_header}; use crate::save::{ClientDb, SaveFlowRequest, SaveFlowResponse, WorkspaceSummary}; -use crate::settings::ProxySettings; +use crate::settings::{HttpVersion, ProxySettings}; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64; use chrono::{DateTime, Utc}; @@ -169,6 +169,7 @@ impl ProxyCtx { .map_err(|e| format!("Failed to load proxy CA: {e}"))?; let mut config = ProxyConfig::new(port.unwrap_or(settings.port), Arc::new(ca)); config.capture_body_limit = settings.capture_body_limit; + config.http2 = settings.http_version == HttpVersion::Auto; // Filled once the engine has bound, since the lookup matches sockets by its port. let attribution_slot: Arc> = Arc::new(OnceLock::new()); let slot = attribution_slot.clone(); @@ -404,6 +405,20 @@ impl ProxyCtx { settings.save(&self.settings_path()) } + /// The HTTP versions offered to clients, saved in `settings.json` and applied to the running + /// engine's next connections. + pub fn set_http_version(&self, version: HttpVersion) -> Result<(), String> { + let mut settings = self.settings()?; + settings.http_version = version; + settings.save(&self.settings_path())?; + if let Ok(handle) = self.handle.lock() + && let Some(handle) = handle.as_ref() + { + handle.set_http2(version == HttpVersion::Auto); + } + Ok(()) + } + /// Intercepts again the hosts passed through after clients rejected the CA, as they no /// longer will once it is trusted. pub fn forget_passthrough(&self) { @@ -646,6 +661,8 @@ pub struct GetProxyStateResponse { pub recording: bool, /// The system proxy turns on as the engine starts. pub system_proxy_at_launch: bool, + /// The HTTP versions offered to clients. + pub http_version: HttpVersion, /// Why the engine last failed to start, until it starts. pub start_error: Option, /// Another process that owns the engine, such as the CLI daemon. While one does, this app @@ -680,6 +697,13 @@ pub struct SetSystemProxyAtLaunchRequest { pub enabled: bool, } +#[derive(Deserialize, TS)] +#[ts(export, export_to = "gen_rpc.ts")] +#[serde(rename_all = "camelCase")] +pub struct SetHttpVersionRequest { + pub http_version: HttpVersion, +} + #[derive(Deserialize, TS)] #[ts(export, export_to = "gen_rpc.ts")] pub struct AppIconRequest { @@ -874,6 +898,7 @@ async fn get_proxy_state( configured_port: settings.port, recording: ctx.recording(), system_proxy_at_launch: settings.system_proxy_at_launch, + http_version: settings.http_version, start_error: ctx.start_error(), other_owner, }) @@ -894,6 +919,11 @@ fn set_system_proxy_at_launch( Ok(req.enabled) } +fn set_http_version(ctx: &ProxyCtx, req: SetHttpVersionRequest) -> Result { + ctx.set_http_version(req.http_version).map_err(rpc_error)?; + Ok(req.http_version) +} + async fn take_over(ctx: ProxyCtx, _req: TakeOverRequest) -> Result { ctx.take_over().await.map_err(rpc_error)?; Ok(true) @@ -1218,12 +1248,27 @@ fn run_event_loop(rx: std::sync::mpsc::Receiver, sink: FlowSink) { write_body_chunk(&mut f.response_body, &data); } } - ProxyEvent::ResponseBodyComplete { id, size, truncated, elapsed_ms, error } => { + ProxyEvent::ResponseBodyComplete { + id, + size, + truncated, + elapsed_ms, + error, + cancelled, + } => { if let Some(mut f) = in_flight.remove(&id) { f.flow.response_size = size as i64; f.flow.response_truncated = truncated; f.flow.elapsed_ms = Some(clamp_ms(elapsed_ms)); f.flow.error = error; + f.flow.cancelled = cancelled; + finish_flow(&db, &events, &attribution, f); + } + } + ProxyEvent::Cancelled { id, elapsed_ms } => { + if let Some(mut f) = in_flight.remove(&id) { + f.flow.elapsed_ms = Some(clamp_ms(elapsed_ms)); + f.flow.cancelled = true; finish_flow(&db, &events, &attribution, f); } } @@ -1306,6 +1351,7 @@ define_rpc! { list_flows(flows::FlowFilter) -> ListFlowsResponse, set_recording(SetRecordingRequest) -> bool, set_system_proxy_at_launch(SetSystemProxyAtLaunchRequest) -> bool, + set_http_version(SetHttpVersionRequest) -> HttpVersion, } async_commands { app_icon(AppIconRequest) -> Option, diff --git a/crates-proxy/yaak-proxy-lib/src/models.rs b/crates-proxy/yaak-proxy-lib/src/models.rs index 5bbd76931..15d35747e 100644 --- a/crates-proxy/yaak-proxy-lib/src/models.rs +++ b/crates-proxy/yaak-proxy-lib/src/models.rs @@ -53,6 +53,9 @@ pub struct Flow { /// Milliseconds from `started_at` to the end of the response. pub elapsed_ms: Option, pub error: Option, + /// The client gave up before the response ended, by resetting its HTTP/2 stream or closing + /// the connection. Not an error. + pub cancelled: bool, /// The id of the Yaak request this flow was last saved as. pub saved_request_id: Option, /// The process that opened the client connection, where the platform can tell (macOS). @@ -155,6 +158,7 @@ impl UpsertModelInfo for Flow { (ElapsedHeadersMs, self.elapsed_headers_ms.into()), (ElapsedMs, self.elapsed_ms.into()), (Error, self.error.into()), + (Cancelled, self.cancelled.into()), (SavedRequestId, self.saved_request_id.into()), (ClientPid, self.client_pid.into()), (ClientName, self.client_name.into()), @@ -183,6 +187,7 @@ impl UpsertModelInfo for Flow { FlowIden::ElapsedHeadersMs, FlowIden::ElapsedMs, FlowIden::Error, + FlowIden::Cancelled, FlowIden::SavedRequestId, FlowIden::ClientPid, FlowIden::ClientName, @@ -218,6 +223,7 @@ impl UpsertModelInfo for Flow { elapsed_headers_ms: r.get("elapsed_headers_ms")?, elapsed_ms: r.get("elapsed_ms")?, error: r.get("error")?, + cancelled: added_column(r, "cancelled")?.unwrap_or_default(), saved_request_id: added_column(r, "saved_request_id")?, client_pid: added_column(r, "client_pid")?, client_name: added_column(r, "client_name")?, diff --git a/crates-proxy/yaak-proxy-lib/src/settings.rs b/crates-proxy/yaak-proxy-lib/src/settings.rs index 981bc3a67..59159ed3f 100644 --- a/crates-proxy/yaak-proxy-lib/src/settings.rs +++ b/crates-proxy/yaak-proxy-lib/src/settings.rs @@ -1,5 +1,6 @@ use serde::{Deserialize, Serialize}; use std::path::Path; +use ts_rs::TS; use yaak_proxy::DEFAULT_CAPTURE_BODY_LIMIT; /// `settings.json` in the proxy data directory. Missing keys take their defaults. @@ -14,6 +15,8 @@ pub struct ProxySettings { pub system_proxy_at_launch: bool, /// Bytes of each request and response body stored per flow. pub capture_body_limit: usize, + /// The HTTP versions offered to clients in intercepted HTTPS. + pub http_version: HttpVersion, } impl Default for ProxySettings { @@ -22,10 +25,24 @@ impl Default for ProxySettings { port: 9090, system_proxy_at_launch: true, capture_body_limit: DEFAULT_CAPTURE_BODY_LIMIT, + http_version: HttpVersion::Auto, } } } +/// The HTTP versions offered to clients in intercepted HTTPS. Upstream, the engine speaks +/// whichever the server picks either way. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export, export_to = "gen_rpc.ts")] +pub enum HttpVersion { + /// HTTP/2 or HTTP/1.1, as the client prefers. + #[default] + Auto, + /// HTTP/1.1 alone, for clients that misbehave on HTTP/2. + Http1Only, +} + impl ProxySettings { /// The settings at `path`, or the defaults when there is no file. pub fn load(path: &Path) -> Result { @@ -70,8 +87,12 @@ mod tests { fn round_trip() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("settings.json"); - let settings = - ProxySettings { port: 8080, system_proxy_at_launch: false, capture_body_limit: 1024 }; + let settings = ProxySettings { + port: 8080, + system_proxy_at_launch: false, + capture_body_limit: 1024, + http_version: HttpVersion::Http1Only, + }; settings.save(&path).unwrap(); assert_eq!(ProxySettings::load(&path).unwrap(), settings); @@ -79,10 +100,25 @@ mod tests { serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap(); assert_eq!( json, - serde_json::json!({"port": 8080, "systemProxyAtLaunch": false, "captureBodyLimit": 1024}) + serde_json::json!({ + "port": 8080, + "systemProxyAtLaunch": false, + "captureBodyLimit": 1024, + "httpVersion": "http1Only", + }) ); } + #[test] + fn http_version_defaults_to_auto() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("settings.json"); + std::fs::write(&path, r#"{"port": 8888}"#).unwrap(); + assert_eq!(ProxySettings::load(&path).unwrap().http_version, HttpVersion::Auto); + std::fs::write(&path, r#"{"httpVersion": "auto"}"#).unwrap(); + assert_eq!(ProxySettings::load(&path).unwrap().http_version, HttpVersion::Auto); + } + #[test] fn the_earlier_system_proxy_key_carries_over() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates-proxy/yaak-proxy-lib/tests/flow.rs b/crates-proxy/yaak-proxy-lib/tests/flow.rs index 3fcf136ab..2c5c2cc09 100644 --- a/crates-proxy/yaak-proxy-lib/tests/flow.rs +++ b/crates-proxy/yaak-proxy-lib/tests/flow.rs @@ -67,6 +67,7 @@ async fn route(req: Request) -> Result>, Infallib .header("content-encoding", "gzip") .body(Full::new(Bytes::new())) .unwrap(), + "/hang" => std::future::pending().await, // XML that only its body says is XML "/xml-as-text" => Response::builder() .header("content-type", "text/plain") @@ -106,8 +107,12 @@ async fn wait_for_flow(ctx: &ProxyCtx, url: &str) -> Flow { #[tokio::test(flavor = "multi_thread")] async fn flow_is_stored_with_its_bodies() { let dir = tempfile::tempdir().unwrap(); - let settings = - ProxySettings { port: 0, capture_body_limit: CAPTURE_LIMIT, system_proxy_at_launch: false }; + let settings = ProxySettings { + port: 0, + capture_body_limit: CAPTURE_LIMIT, + system_proxy_at_launch: false, + ..Default::default() + }; settings.save(&dir.path().join("settings.json")).unwrap(); let (events, _event_rx) = RpcEventEmitter::new(); @@ -174,8 +179,12 @@ async fn flow_is_stored_with_its_bodies() { #[tokio::test(flavor = "multi_thread")] async fn early_response_keeps_request_totals() { let dir = tempfile::tempdir().unwrap(); - let settings = - ProxySettings { port: 0, capture_body_limit: CAPTURE_LIMIT, system_proxy_at_launch: false }; + let settings = ProxySettings { + port: 0, + capture_body_limit: CAPTURE_LIMIT, + system_proxy_at_launch: false, + ..Default::default() + }; settings.save(&dir.path().join("settings.json")).unwrap(); let (events, _event_rx) = RpcEventEmitter::new(); @@ -204,11 +213,84 @@ async fn rpc(ctx: &ProxyCtx, cmd: &str, payload: Value) -> Result yaak_proxy_lib::build_router().dispatch(cmd, payload, ctx).await.map_err(|e| e.message) } +/// A request the client gives up on is stored as cancelled, not as an error. +#[tokio::test(flavor = "multi_thread")] +async fn cancelled_flow_is_not_an_error() { + let dir = tempfile::tempdir().unwrap(); + let settings = ProxySettings { port: 0, system_proxy_at_launch: false, ..Default::default() }; + settings.save(&dir.path().join("settings.json")).unwrap(); + + let (events, _event_rx) = RpcEventEmitter::new(); + let ctx = ProxyCtx::new(dir.path(), events).unwrap(); + ctx.start().unwrap(); + let port = ctx.port().unwrap(); + + let upstream = spawn_upstream().await; + let client = reqwest::Client::builder() + .proxy(reqwest::Proxy::all(format!("http://127.0.0.1:{port}")).unwrap()) + .timeout(Duration::from_millis(300)) + .build() + .unwrap(); + let url = format!("http://{upstream}/hang"); + assert!(client.get(&url).send().await.unwrap_err().is_timeout()); + + let flow = wait_for_flow(&ctx, &url).await; + assert!(flow.cancelled); + assert_eq!((flow.error, flow.status), (None, None)); + assert!(flow.elapsed_ms.is_some()); +} + +/// `http1Only` is saved and reaches the running engine: the next connection from a client that +/// offers HTTP/2 speaks HTTP/1.1, and `auto` brings HTTP/2 back. +#[tokio::test(flavor = "multi_thread")] +async fn http_version_setting_applies_to_new_connections() { + let dir = tempfile::tempdir().unwrap(); + let settings = ProxySettings { port: 0, system_proxy_at_launch: false, ..Default::default() }; + settings.save(&dir.path().join("settings.json")).unwrap(); + + let (events, _event_rx) = RpcEventEmitter::new(); + let ctx = ProxyCtx::new(dir.path(), events).unwrap(); + ctx.start().unwrap(); + let port = ctx.port().unwrap(); + let ca = std::fs::read(dir.path().join("ca/ca.pem")).unwrap(); + // Nothing listens there, so each flow is a 502, but its client side is spoken all the same. + let closed = TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap().port(); + + let mut sent = 0; + let mut client_version = async || { + sent += 1; + // A client of its own, so a connection of its own. + let client = reqwest::Client::builder() + .proxy(reqwest::Proxy::all(format!("http://127.0.0.1:{port}")).unwrap()) + .add_root_certificate(reqwest::Certificate::from_pem(&ca).unwrap()) + .build() + .unwrap(); + let url = format!("https://localhost:{closed}/{sent}"); + assert_eq!(client.get(&url).send().await.unwrap().status(), 502); + wait_for_flow(&ctx, &url).await.client_http_version + }; + + assert_eq!(client_version().await, "HTTP/2"); + let set = rpc(&ctx, "set_http_version", json!({"httpVersion": "http1Only"})).await.unwrap(); + assert_eq!(set, json!("http1Only")); + assert_eq!(rpc(&ctx, "get_proxy_state", json!({})).await.unwrap()["httpVersion"], "http1Only"); + assert_eq!(client_version().await, "HTTP/1.1"); + + rpc(&ctx, "set_http_version", json!({"httpVersion": "auto"})).await.unwrap(); + assert_eq!(client_version().await, "HTTP/2"); + let saved = ProxySettings::load(&dir.path().join("settings.json")).unwrap(); + assert_eq!(saved.http_version, yaak_proxy_lib::settings::HttpVersion::Auto); +} + #[tokio::test(flavor = "multi_thread")] async fn body_rpcs() { let dir = tempfile::tempdir().unwrap(); - let settings = - ProxySettings { port: 0, capture_body_limit: CAPTURE_LIMIT, system_proxy_at_launch: false }; + let settings = ProxySettings { + port: 0, + capture_body_limit: CAPTURE_LIMIT, + system_proxy_at_launch: false, + ..Default::default() + }; settings.save(&dir.path().join("settings.json")).unwrap(); let (events, _event_rx) = RpcEventEmitter::new(); @@ -225,6 +307,7 @@ async fn body_rpcs() { "configuredPort": 0, "recording": true, "systemProxyAtLaunch": false, + "httpVersion": "auto", "startError": null, "otherOwner": null, }) diff --git a/crates/yaak-proxy/Cargo.toml b/crates/yaak-proxy/Cargo.toml index 58a52bc50..aa055f251 100644 --- a/crates/yaak-proxy/Cargo.toml +++ b/crates/yaak-proxy/Cargo.toml @@ -6,7 +6,7 @@ publish = false [dependencies] hyper = { version = "1", features = ["http1", "http2", "server", "client"] } -hyper-util = { version = "0.1", features = ["tokio", "http1", "client-legacy"] } +hyper-util = { version = "0.1", features = ["tokio", "http1", "client-legacy", "server-auto"] } http-body-util = "0.1" http = "1" bytes = "1" diff --git a/crates/yaak-proxy/src/body.rs b/crates/yaak-proxy/src/body.rs index 13b79fc34..cc814525f 100644 --- a/crates/yaak-proxy/src/body.rs +++ b/crates/yaak-proxy/src/body.rs @@ -1,5 +1,6 @@ use std::pin::Pin; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Poll}; use std::time::Instant; @@ -25,6 +26,8 @@ pub(crate) struct CaptureBody { captured: usize, size: u64, finished: bool, + /// Set when reading the body fails. + failed: Option>, } impl CaptureBody { @@ -35,7 +38,23 @@ impl CaptureBody { start: Instant, engine: Arc, ) -> Self { - Self { inner, direction, id, start, engine, captured: 0, size: 0, finished: false } + Self { + inner, + direction, + id, + start, + engine, + captured: 0, + size: 0, + finished: false, + failed: None, + } + } + + /// Sets `failed` if reading the body fails, as a client's upload does when it cancels. + pub(crate) fn flag_failure(mut self, failed: Arc) -> Self { + self.failed = Some(failed); + self } /// Every byte the body declared has arrived. An HTTP/2 stream's end can come in a frame after @@ -45,7 +64,7 @@ impl CaptureBody { self.inner.is_end_stream() || self.inner.size_hint().exact() == Some(0) } - fn finish(&mut self, error: Option) { + fn finish(&mut self, error: Option, cancelled: bool) { if self.finished { return; } @@ -62,6 +81,7 @@ impl CaptureBody { truncated, elapsed_ms: self.start.elapsed().as_millis() as u64, error, + cancelled, }, }; self.engine.emit(event); @@ -92,14 +112,17 @@ impl Body for CaptureBody { }); } if self.complete() { - self.finish(None); + self.finish(None, false); } } Poll::Ready(Some(Err(e))) => { + if let Some(failed) = &self.failed { + failed.store(true, Ordering::Relaxed); + } let error = e.to_string(); - self.finish(Some(error)); + self.finish(Some(error), false); } - Poll::Ready(None) => self.finish(None), + Poll::Ready(None) => self.finish(None, false), Poll::Pending => {} } result @@ -114,10 +137,11 @@ impl Body for CaptureBody { } } +/// A response body dropped before its end was abandoned by the client, which reset its stream or +/// closed the connection. Dropping the upstream body with it cancels the request there too. impl Drop for CaptureBody { fn drop(&mut self) { - let error = (self.direction == Direction::Response && !self.complete()) - .then(|| "Client closed the connection before the response finished".to_string()); - self.finish(error); + let cancelled = self.direction == Direction::Response && !self.complete(); + self.finish(None, cancelled); } } diff --git a/crates/yaak-proxy/src/cert.rs b/crates/yaak-proxy/src/cert.rs index cfdd73a44..a3674ef35 100644 --- a/crates/yaak-proxy/src/cert.rs +++ b/crates/yaak-proxy/src/cert.rs @@ -24,7 +24,8 @@ pub struct CaStore { cert_der: CertificateDer<'static>, issuer: Certificate, key: KeyPair, - leaf_configs: Mutex>>, + /// Keyed by host and whether HTTP/2 is offered. + leaf_configs: Mutex>>, } impl CaStore { @@ -92,14 +93,50 @@ impl CaStore { &self.key_path } + /// The TLS config presented to clients of `host`, offering `h2` in ALPN when `http2` is set + /// and `http/1.1` always. pub(crate) fn server_config( &self, host: &str, - ) -> Result, Box> { - if let Some(config) = self.leaf_configs.lock().unwrap().get(host) { - return Ok(config.clone()); + http2: bool, + ) -> Result, BoxError> { + self.server_config_with(host, http2, || self.sign_leaf(host)) + } + + /// [`Self::server_config`], with `sign` making a new leaf. It runs outside the cache's lock, + /// so a host being signed holds up no other host's lookup. Connections racing to the same + /// new host may each sign one; the first stored is the one kept. + fn server_config_with( + &self, + host: &str, + http2: bool, + sign: impl FnOnce() -> Result, + ) -> Result, BoxError> { + let key = (host.to_string(), http2); + { + let mut configs = self.leaf_configs.lock().unwrap(); + if let Some(config) = configs.get(&key) { + return Ok(config.clone()); + } + // The same leaf serves both, so switching HTTP/2 off and on signs nothing new. + if let Some(other) = configs.get(&(host.to_string(), !http2)) { + let mut config = ServerConfig::clone(other); + config.alpn_protocols = alpn(http2); + let config = Arc::new(config); + configs.insert(key, config.clone()); + return Ok(config); + } } + let mut config = sign()?; + config.alpn_protocols = alpn(http2); + let mut configs = self.leaf_configs.lock().unwrap(); + Ok(configs.entry(key).or_insert_with(|| Arc::new(config)).clone()) + } + + /// A TLS config with a new leaf for `host`, signed by the CA. Key generation and signing + /// take a few milliseconds. + fn sign_leaf(&self, host: &str) -> Result { let now = OffsetDateTime::now_utc(); let mut params = CertificateParams::new(vec![host.to_string()])?; params.distinguished_name.push(DnType::CommonName, host); @@ -115,22 +152,22 @@ impl CaStore { let leaf_key = KeyPair::generate()?; let leaf_cert = params.signed_by(&leaf_key, &self.issuer, &self.key)?; - let mut config = - ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) - .with_safe_default_protocol_versions()? - .with_no_client_auth() - .with_single_cert( - vec![leaf_cert.der().clone(), self.cert_der.clone()], - PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(leaf_key.serialize_der())), - )?; - config.alpn_protocols = vec![b"http/1.1".to_vec()]; - - let config = Arc::new(config); - self.leaf_configs.lock().unwrap().insert(host.to_string(), config.clone()); - Ok(config) + Ok(ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions()? + .with_no_client_auth() + .with_single_cert( + vec![leaf_cert.der().clone(), self.cert_der.clone()], + PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(leaf_key.serialize_der())), + )?) } } +type BoxError = Box; + +fn alpn(http2: bool) -> Vec> { + if http2 { vec![b"h2".to_vec(), b"http/1.1".to_vec()] } else { vec![b"http/1.1".to_vec()] } +} + /// Issuer fields for the CA. Leaf certificates are signed against these, so they must stay /// identical to what was written to `ca.pem`. fn ca_params() -> CertificateParams { @@ -173,3 +210,45 @@ fn write_atomic(path: &Path, contents: &[u8], private: bool) -> io::Result<()> { file.sync_all()?; fs::rename(&tmp, path) } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + + /// A host whose leaf is still being signed holds up no lookup of another host's. + #[test] + fn signing_one_host_does_not_block_another() { + let dir = tempfile::tempdir().unwrap(); + let ca = Arc::new(CaStore::load_or_create(dir.path()).unwrap()); + let cached = ca.server_config("cached.test", true).unwrap(); + + let (signing_tx, signing_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let slow = std::thread::spawn({ + let ca = ca.clone(); + move || { + ca.server_config_with("slow.test", true, || { + signing_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + ca.sign_leaf("slow.test") + }) + } + }); + signing_rx.recv().unwrap(); + + let (found_tx, found_rx) = mpsc::channel(); + std::thread::spawn({ + let ca = ca.clone(); + move || found_tx.send(ca.server_config("cached.test", true).unwrap()) + }); + let found = found_rx.recv_timeout(std::time::Duration::from_secs(5)); + release_tx.send(()).unwrap(); + let found = found.expect("the cached lookup waited for another host's signing"); + assert!(Arc::ptr_eq(&found, &cached)); + + let slow = slow.join().unwrap().unwrap(); + assert_eq!(slow.alpn_protocols, alpn(true)); + assert!(Arc::ptr_eq(&slow, &ca.server_config("slow.test", true).unwrap()), "kept"); + } +} diff --git a/crates/yaak-proxy/src/connection.rs b/crates/yaak-proxy/src/connection.rs index c950909dd..6498c72ee 100644 --- a/crates/yaak-proxy/src/connection.rs +++ b/crates/yaak-proxy/src/connection.rs @@ -2,13 +2,30 @@ use std::sync::Arc; use hyper::server::conn::http1; use hyper::service::service_fn; -use hyper_util::rt::TokioIo; +use hyper_util::rt::{TokioExecutor, TokioIo}; +use hyper_util::server::conn::auto; use tokio::io::{AsyncRead, AsyncWrite}; use crate::request::handle_request; use crate::{ClientConn, Engine}; -pub(crate) async fn serve_http1(io: S, handler: F) -> Result<(), hyper::Error> +/// Streams one HTTP/2 client connection may have open at once. +const MAX_CONCURRENT_STREAMS: u32 = 100; +/// What an HTTP/2 client may send before the engine has forwarded it, per stream and in all. +/// The same as hyper's client uses upstream, so uploads are not throttled in between. +const STREAM_WINDOW: u32 = 2 * 1024 * 1024; +const CONNECTION_WINDOW: u32 = 5 * 1024 * 1024; +/// Request headers an HTTP/2 client may send, cookies included. hyper's 16 KB default is less +/// than a browser sends some sites. +const MAX_HEADER_LIST_SIZE: u32 = 256 * 1024; + +/// Serves a decrypted client connection in the protocol it chose in ALPN: HTTP/2, with each +/// stream a request of its own, or HTTP/1.1 as on the plain listener. +pub(crate) async fn serve_tls( + io: S, + http2: bool, + handler: F, +) -> Result<(), Box> where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, F: Fn(hyper::Request) -> Fut + Send + Sync + 'static, @@ -16,18 +33,31 @@ where + Send + 'static, { - http1::Builder::new() - .preserve_header_case(true) - .title_case_headers(true) - .serve_connection(TokioIo::new(io), service_fn(handler)) - .with_upgrades() - .await + let mut builder = auto::Builder::new(TokioExecutor::new()); + builder.http1().preserve_header_case(true).title_case_headers(true); + builder + .http2() + .max_concurrent_streams(MAX_CONCURRENT_STREAMS) + .initial_stream_window_size(STREAM_WINDOW) + .initial_connection_window_size(CONNECTION_WINDOW) + .max_header_list_size(MAX_HEADER_LIST_SIZE); + let builder = if http2 { builder.http2_only() } else { builder.http1_only() }; + builder.serve_connection_with_upgrades(TokioIo::new(io), service_fn(handler)).await } +/// Serves a connection to the listener, in HTTP/1.1 only: HTTP/2 is offered inside TLS alone. pub(crate) async fn handle_connection( stream: tokio::net::TcpStream, client: ClientConn, engine: Arc, ) -> Result<(), hyper::Error> { - serve_http1(stream, move |req| handle_request(req, client, engine.clone())).await + http1::Builder::new() + .preserve_header_case(true) + .title_case_headers(true) + .serve_connection( + TokioIo::new(stream), + service_fn(move |req| handle_request(req, client, engine.clone())), + ) + .with_upgrades() + .await } diff --git a/crates/yaak-proxy/src/lib.rs b/crates/yaak-proxy/src/lib.rs index 5ac8646d7..f7401932e 100644 --- a/crates/yaak-proxy/src/lib.rs +++ b/crates/yaak-proxy/src/lib.rs @@ -8,7 +8,7 @@ use std::collections::hash_map::RandomState; use std::collections::{HashMap, HashSet}; use std::hash::{BuildHasher, Hasher}; use std::net::SocketAddr; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::mpsc as std_mpsc; use std::sync::{Arc, Mutex}; use std::time::SystemTime; @@ -46,6 +46,9 @@ pub struct ProxyConfig { /// number and the client's address, before anything is read from it, so the client is /// still there to be looked up. The connection waits for it. pub on_accept: Option, + /// Offer HTTP/2 to clients in intercepted TLS, beside HTTP/1.1. Off, they all speak + /// HTTP/1.1, for clients that misbehave on HTTP/2. See [`ProxyHandle::set_http2`]. + pub http2: bool, } /// See [`ProxyConfig::on_accept`]. @@ -59,14 +62,16 @@ impl ProxyConfig { capture_body_limit: DEFAULT_CAPTURE_BODY_LIMIT, upstream_root_certs: Vec::new(), on_accept: None, + http2: true, } } } /// Lifecycle of each proxied exchange (a flow), keyed by `id`. A flow starts with -/// `RequestStart` and ends with exactly one `ResponseBodyComplete` or `Error`. Traffic the -/// engine tunnels without decrypting (pinned clients, non-TLS CONNECT payloads, the frames -/// after a `101 Switching Protocols`) produces no events. +/// `RequestStart` and ends with exactly one `ResponseBodyComplete`, `Error` or `Cancelled`. +/// Each HTTP/2 stream is a flow of its own. Traffic the engine tunnels without decrypting +/// (pinned clients, non-TLS CONNECT payloads, the frames after a `101 Switching Protocols`) +/// produces no events. #[derive(Debug, Clone)] pub enum ProxyEvent { /// Request headers arrived from the client. @@ -125,7 +130,7 @@ pub enum ProxyEvent { /// The captured part of the frame; empty once `capture_body_limit` is reached. data: Bytes, }, - /// The response body ended, normally or with `error` (upstream failed or client left). + /// The response body ended: normally, with `error` when upstream failed, or `cancelled`. ResponseBodyComplete { id: u64, size: u64, @@ -134,6 +139,9 @@ pub enum ProxyEvent { /// Milliseconds from `started_at` to the end of the body. elapsed_ms: u64, error: Option, + /// The client stopped reading before the body ended: it reset the stream or closed the + /// connection. Not an error of the engine's or the server's. + cancelled: bool, }, /// No response: connecting to or requesting from upstream failed, and the client got a 502, /// or the engine refused the request (no host to send it to, or it looped back to the engine) @@ -143,6 +151,9 @@ pub enum ProxyEvent { error: String, elapsed_ms: u64, }, + /// The client gave up before the response headers arrived, by resetting the stream or + /// closing the connection. The request upstream is cancelled with it. + Cancelled { id: u64, elapsed_ms: u64 }, } /// A running engine. Dropping it stops the listener and joins its thread. @@ -168,6 +179,12 @@ impl ProxyHandle { self.engine.handshake_failures.lock().unwrap().clear(); } + /// Offers HTTP/2 to clients, or stops, from the next connection on. Connections already + /// open keep the protocol they chose. + pub fn set_http2(&self, enabled: bool) { + self.engine.http2.store(enabled, Ordering::Relaxed); + } + /// Handshake failures counted toward passing `addr` (`host:port`) through, and whether it /// already is. For tests that must wait for the engine to record a failure. #[doc(hidden)] @@ -206,6 +223,8 @@ pub(crate) struct Engine { upstream_tls_h1: Arc, passthrough: Mutex>, handshake_failures: Mutex>, + /// Offer HTTP/2 to clients in intercepted TLS. + http2: AtomicBool, /// `Via` value marking requests whose URL the engine rebuilt from `Host`, unique to this run, /// so one that comes back to the engine is refused instead of forwarded again. via: HeaderValue, @@ -220,6 +239,10 @@ impl Engine { self.next_id.fetch_add(1, Ordering::Relaxed) } + fn offers_http2(&self) -> bool { + self.http2.load(Ordering::Relaxed) + } + fn is_passthrough(&self, addr: &str) -> bool { self.passthrough.lock().unwrap().contains(addr) } @@ -284,6 +307,7 @@ pub fn start_proxy(config: ProxyConfig) -> Result { upstream_tls_h1: upstream_tls_config(&roots, &[b"http/1.1"])?, passthrough: Mutex::new(HashSet::new()), handshake_failures: Mutex::new(HashMap::new()), + http2: AtomicBool::new(config.http2), via: via_marker(), }); diff --git a/crates/yaak-proxy/src/request.rs b/crates/yaak-proxy/src/request.rs index 7427b0750..6b76ab3ff 100644 --- a/crates/yaak-proxy/src/request.rs +++ b/crates/yaak-proxy/src/request.rs @@ -2,12 +2,18 @@ use std::convert::Infallible; use std::error::Error; use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant, SystemTime}; use bytes::Bytes; use http_body_util::{BodyExt, Full}; +use hyper::body::Body; use hyper::body::Incoming; -use hyper::header::{CONTENT_TYPE, HOST, UPGRADE, VIA}; +use hyper::client::conn::TrySendError; +use hyper::header::{ + CONTENT_LENGTH, CONTENT_TYPE, COOKIE, HOST, HeaderValue, TRANSFER_ENCODING, UPGRADE, VIA, +}; +use hyper::http::request::Parts; use hyper::http::uri::{Authority, PathAndQuery, Scheme}; use hyper::upgrade::OnUpgrade; use hyper::{Method, Request, Response, StatusCode, Uri}; @@ -20,7 +26,7 @@ use tokio::net::TcpStream; use tokio_rustls::{LazyConfigAcceptor, TlsConnector}; use crate::body::{CaptureBody, Direction}; -use crate::connection::serve_http1; +use crate::connection::serve_tls; use crate::tunnel::{Rewind, splice, tunnel}; use crate::{ClientConn, Engine, ProxyEvent}; @@ -64,6 +70,10 @@ struct Flow { id: u64, start: Instant, engine: Arc, + /// It has answered the client, so dropping it is no cancel. + answered: bool, + /// Reading the client's upload failed: it reset the stream or closed the connection. + upload_failed: Arc, } impl Flow { @@ -73,7 +83,13 @@ impl Flow { req: &Request, url: String, ) -> Self { - let flow = Self { id: engine.next_flow_id(), start: Instant::now(), engine }; + let flow = Self { + id: engine.next_flow_id(), + start: Instant::now(), + engine, + answered: false, + upload_failed: Arc::new(AtomicBool::new(false)), + }; flow.engine.emit(ProxyEvent::RequestStart { id: flow.id, started_at: SystemTime::now(), @@ -83,6 +99,18 @@ impl Flow { connection: client.connection, client_addr: client.addr, }); + // HTTP/2 carries the host in `:authority`, which hyper keeps in the URI. Shown as `Host`, + // as HTTP/1.1 requests have it. + if req.version() == hyper::Version::HTTP_2 + && !req.headers().contains_key(HOST) + && let Some(authority) = req.uri().authority() + { + flow.engine.emit(ProxyEvent::RequestHeader { + id: flow.id, + name: HOST.to_string(), + value: authority.to_string(), + }); + } for (name, value) in req.headers() { flow.engine.emit(ProxyEvent::RequestHeader { id: flow.id, @@ -94,14 +122,17 @@ impl Flow { } fn capture_request(&self, body: Incoming) -> BoxBody { - CaptureBody::new(body, Direction::Request, self.id, self.start, self.engine.clone()).boxed() + CaptureBody::new(body, Direction::Request, self.id, self.start, self.engine.clone()) + .flag_failure(self.upload_failed.clone()) + .boxed() } fn respond( - self, + mut self, resp: Response, remote_addr: Option, ) -> Response { + self.answered = true; self.engine.emit(ProxyEvent::ResponseStart { id: self.id, status: resp.status().as_u16(), @@ -117,12 +148,14 @@ impl Flow { }); } let (parts, body) = resp.into_parts(); - let body = CaptureBody::new(body, Direction::Response, self.id, self.start, self.engine); + let engine = self.engine.clone(); + let body = CaptureBody::new(body, Direction::Response, self.id, self.start, engine); Response::from_parts(parts, body.boxed()) } /// Answers the client with `status` and `error` without contacting any server. - fn reject(self, status: StatusCode, error: &str) -> Response { + fn reject(mut self, status: StatusCode, error: &str) -> Response { + self.answered = true; debug!("Flow {} rejected: {error}", self.id); self.engine.emit(ProxyEvent::Error { id: self.id, @@ -136,7 +169,14 @@ impl Flow { .unwrap() } - fn fail(self, error: String) -> Response { + fn fail(mut self, error: String) -> Response { + // The request failed because the client stopped sending its body: it cancelled. The + // answer goes nowhere, and dropping the flow unanswered says so. + if self.upload_failed.load(Ordering::Relaxed) { + debug!("Flow {} upload ended by the client: {error}", self.id); + return bad_gateway(error); + } + self.answered = true; debug!("Flow {} failed: {error}", self.id); let body = format!("Yaak Proxy could not reach the server: {error}"); self.engine.emit(ProxyEvent::Error { @@ -148,6 +188,20 @@ impl Flow { } } +/// A flow dropped before it answered was abandoned by the client, which reset its stream or +/// closed the connection: hyper drops the request's future, and the request upstream with it. +impl Drop for Flow { + fn drop(&mut self) { + if !self.answered { + debug!("Flow {} cancelled by the client", self.id); + self.engine.emit(ProxyEvent::Cancelled { + id: self.id, + elapsed_ms: self.start.elapsed().as_millis() as u64, + }); + } + } +} + fn bad_gateway(body: String) -> Response { Response::builder().status(StatusCode::BAD_GATEWAY).body(text_body(body)).unwrap() } @@ -296,7 +350,7 @@ where .server_name() .map(str::to_string) .unwrap_or_else(|| target.server_name.clone()); - let config = engine.ca.server_config(&server_name)?; + let config = engine.ca.server_config(&server_name, engine.offers_http2())?; let mut tls = match start.into_stream(config).await { Ok(tls) => tls, Err(e) => { @@ -323,10 +377,11 @@ where buf.truncate(n); engine.record_handshake_success(&target.addr); + let http2 = tls.get_ref().1.alpn_protocol() == Some(b"h2"); let tls = Rewind::new(Bytes::from(buf), tls); - let target = Arc::new(Target { server_name, ..target }); - serve_http1(tls, move |req| forward_https(req, target.clone(), conn, engine.clone())).await?; - Ok(()) + let upstreams = Arc::new(Upstreams::new(Target { server_name, ..target })); + serve_tls(tls, http2, move |req| forward_https(req, upstreams.clone(), conn, engine.clone())) + .await } fn record_handshake_failure( @@ -371,11 +426,115 @@ enum Upstream { H2(hyper::client::conn::http2::SendRequest), } +/// A connection to the server, and the address it reached. +struct UpstreamConn { + sender: Upstream, + remote_addr: Option, +} + +impl UpstreamConn { + fn is_h2(&self) -> bool { + matches!(self.sender, Upstream::H2(_)) + } + + fn is_closed(&self) -> bool { + match &self.sender { + Upstream::H1(s) => s.is_closed(), + Upstream::H2(s) => s.is_closed(), + } + } + + /// Can take a request now: an HTTP/2 connection any time it is open, an HTTP/1.1 one once + /// the response before has been read. + fn is_ready(&self) -> bool { + match &self.sender { + Upstream::H1(s) => s.is_ready(), + Upstream::H2(s) => s.is_ready(), + } + } + + async fn send( + &mut self, + req: Request, + ) -> Result, TrySendError>> { + match &mut self.sender { + Upstream::H1(s) => s.try_send_request(req).await, + Upstream::H2(s) => s.try_send_request(req).await, + } + } +} + +/// Idle HTTP/1.1 connections kept per client connection, for its requests to come. +const MAX_IDLE_H1: usize = 6; + +/// The connections to the server of one intercepted client connection, reused by its requests, +/// so an HTTP/2 client's streams do not each pay for a handshake. An HTTP/2 connection carries +/// them all at once; an HTTP/1.1 one a request at a time. +struct Upstreams { + target: Target, + conns: tokio::sync::Mutex>, + /// The server answered in HTTP/1.1, so each request in flight needs a connection of its + /// own, and new ones need not wait to see whether the one before can be shared. + h1: AtomicBool, +} + +impl Upstreams { + fn new(target: Target) -> Self { + Self { target, conns: tokio::sync::Mutex::new(Vec::new()), h1: AtomicBool::new(false) } + } + + /// A connection for one request, and whether it carried one before. Requests that arrive + /// before the first connection is up wait for it, to share it if it is HTTP/2. + async fn take(&self, engine: &Engine) -> Result<(UpstreamConn, bool), BoxError> { + let mut conns = self.conns.lock().await; + conns.retain(|c| !c.is_closed()); + if let Some(i) = conns.iter().position(UpstreamConn::is_ready) { + let conn = match &conns[i].sender { + Upstream::H2(s) => UpstreamConn { + sender: Upstream::H2(s.clone()), + remote_addr: conns[i].remote_addr, + }, + Upstream::H1(_) => conns.swap_remove(i), + }; + return Ok((conn, true)); + } + if self.h1.load(Ordering::Relaxed) { + drop(conns); + return Ok((connect_upstream(engine, &self.target, false).await?, false)); + } + let conn = connect_upstream(engine, &self.target, false).await?; + match &conn.sender { + Upstream::H2(s) => conns.push(UpstreamConn { + sender: Upstream::H2(s.clone()), + remote_addr: conn.remote_addr, + }), + Upstream::H1(_) => self.h1.store(true, Ordering::Relaxed), + } + Ok((conn, false)) + } + + /// Keeps a connection for later requests. An HTTP/1.1 one takes the next once this response + /// has been read. An HTTP/2 one is kept already, unless it was opened to send a request again. + async fn put_back(&self, conn: UpstreamConn) { + if conn.is_closed() { + return; + } + let mut conns = self.conns.lock().await; + if conn.is_h2() { + if !conns.iter().any(|c| c.is_h2() && !c.is_closed()) { + conns.push(conn); + } + } else if conns.len() < MAX_IDLE_H1 { + conns.push(conn); + } + } +} + async fn connect_upstream( engine: &Engine, target: &Target, h1_only: bool, -) -> Result<(Upstream, Option), BoxError> { +) -> Result { let tcp = TcpStream::connect(&target.addr).await?; let remote_addr = tcp.peer_addr().ok(); let config = if h1_only { &engine.upstream_tls_h1 } else { &engine.upstream_tls }; @@ -406,61 +565,118 @@ async fn connect_upstream( }); Upstream::H1(sender) }; - Ok((upstream, remote_addr)) + Ok(UpstreamConn { sender: upstream, remote_addr }) +} + +/// Addresses a request to the server in the protocol of `conn`. It may be addressed for the +/// other one already, when it is sent again on a new connection. +fn address( + parts: &mut Parts, + conn: &UpstreamConn, + target: &Target, + path: &str, + has_body: bool, +) -> Result<(), String> { + // The host the client asked for: `Host` over HTTP/1.1, `:authority` over HTTP/2, which + // hyper keeps in the URI. + let authority = + parts.uri.authority().map_or_else(|| target.authority.clone(), |a| a.to_string()); + if conn.is_h2() { + // `Host` moves into `:authority`. Google's servers reset any stream that has both, even + // when they match. + let host = match parts.headers.remove(HOST) { + Some(host) => host.to_str().map_err(|e| e.to_string())?.to_string(), + None => authority, + }; + parts.uri = format!("https://{host}{path}").parse::().map_err(|e| e.to_string())?; + } else { + parts.uri = path.parse::().map_err(|e| e.to_string())?; + if !parts.headers.contains_key(HOST) + && let Ok(host) = authority.parse() + { + parts.headers.insert(HOST, host); + } + if parts.version == hyper::Version::HTTP_2 { + parts.version = hyper::Version::HTTP_11; + // HTTP/2 frames a body without a length. HTTP/1.1 needs one or chunks, and hyper + // sends a GET with neither as empty. Chunks also carry trailers, those named in + // `Trailer`: hyper's HTTP/1.1 encoder drops any others. + if has_body && !parts.headers.contains_key(CONTENT_LENGTH) { + parts.headers.insert(TRANSFER_ENCODING, HeaderValue::from_static("chunked")); + } + // HTTP/2 may split `Cookie` into a field per cookie; HTTP/1.1 has one (RFC 9113 + // 8.2.3). + if parts.headers.get_all(COOKIE).iter().nth(1).is_some() { + let joined = parts + .headers + .get_all(COOKIE) + .iter() + .map(HeaderValue::as_bytes) + .collect::>() + .join(&b"; "[..]); + let joined = HeaderValue::from_bytes(&joined).map_err(|e| e.to_string())?; + parts.headers.insert(COOKIE, joined); + } + } + } + Ok(()) } async fn forward_https( mut req: Request, - target: Arc, + upstreams: Arc, client: ClientConn, engine: Arc, ) -> Result, Infallible> { + let target = &upstreams.target; let path = req.uri().path_and_query().map(|pq| pq.to_string()).unwrap_or_else(|| "/".into()); let url = format!("https://{}{path}", target.authority); - let flow = Flow::begin(engine.clone(), client, &req, url.clone()); + let flow = Flow::begin(engine.clone(), client, &req, url); + // Only HTTP/1.1 has `Upgrade`. Its connection is the tunnel afterwards, so it is not shared. let client_upgrade: Option = req.headers().contains_key(UPGRADE).then(|| hyper::upgrade::on(&mut req)); - - let (mut upstream, remote_addr) = - match connect_upstream(&engine, &target, client_upgrade.is_some()).await { - Ok(u) => u, - Err(e) => return Ok(flow.fail(error_chain(e.as_ref()))), - }; + let taken = match client_upgrade { + Some(_) => connect_upstream(&engine, target, true).await.map(|c| (c, false)), + None => upstreams.take(&engine).await, + }; + let (mut conn, mut reused) = match taken { + Ok(taken) => taken, + Err(e) => return Ok(flow.fail(error_chain(e.as_ref()))), + }; let (mut parts, body) = req.into_parts(); - let uri = match &upstream { - // `Host` moves into `:authority`. Google's servers reset any stream that has both, even - // when they match. - Upstream::H2(_) => match parts.headers.remove(HOST) { - Some(host) => match host.to_str() { - Ok(host) => format!("https://{host}{path}").parse::(), - Err(e) => return Ok(flow.fail(e.to_string())), - }, - None => url.parse::(), - }, - Upstream::H1(_) => path.parse::(), - }; - parts.uri = match uri { - Ok(uri) => uri, - Err(e) => return Ok(flow.fail(e.to_string())), - }; - if matches!(upstream, Upstream::H1(_)) - && !parts.headers.contains_key(HOST) - && let Ok(host) = target.authority.parse() - { - parts.headers.insert(HOST, host); + if let Err(e) = address(&mut parts, &conn, target, &path, !body.is_end_stream()) { + return Ok(flow.fail(e)); } - let outgoing = Request::from_parts(parts, flow.capture_request(body)); + let mut outgoing = Request::from_parts(parts, flow.capture_request(body)); - let sent = match &mut upstream { - Upstream::H1(s) => s.send_request(outgoing).await, - Upstream::H2(s) => s.send_request(outgoing).await, - }; - let mut resp = match sent { - Ok(resp) => resp, - Err(e) => return Ok(flow.fail(error_chain(&e))), + let mut resp = loop { + let mut e = match conn.send(outgoing).await { + Ok(resp) => break resp, + Err(e) => e, + }; + // A reused connection the server closed as the request went out: the request never + // left, so it goes again, once, on a new connection. + let Some(unsent) = e.take_message().filter(|_| reused) else { + return Ok(flow.fail(error_chain(e.error()))); + }; + debug!("Upstream connection to {} was closed, retrying on a new one", target.addr); + conn = match connect_upstream(&engine, target, false).await { + Ok(conn) => conn, + Err(e) => return Ok(flow.fail(error_chain(e.as_ref()))), + }; + reused = false; + let (mut parts, body) = unsent.into_parts(); + if let Err(e) = address(&mut parts, &conn, target, &path, !body.is_end_stream()) { + return Ok(flow.fail(e)); + } + outgoing = Request::from_parts(parts, body); }; + let remote_addr = conn.remote_addr; + if client_upgrade.is_none() { + upstreams.put_back(conn).await; + } if resp.status() == StatusCode::SWITCHING_PROTOCOLS && let Some(client_upgrade) = client_upgrade diff --git a/crates/yaak-proxy/tests/proxy.rs b/crates/yaak-proxy/tests/proxy.rs index 94bd7d7f9..2e4062ba2 100644 --- a/crates/yaak-proxy/tests/proxy.rs +++ b/crates/yaak-proxy/tests/proxy.rs @@ -1,6 +1,8 @@ +use std::collections::{HashMap, HashSet}; use std::convert::Infallible; use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc::Receiver; use std::time::{Duration, SystemTime}; @@ -9,9 +11,10 @@ use futures_util::{SinkExt, StreamExt}; use http_body_util::combinators::BoxBody; use http_body_util::{BodyExt, Full, StreamBody}; use hyper::body::{Frame, Incoming}; -use hyper::header::{CONNECTION, CONTENT_LENGTH, CONTENT_TYPE, HOST, UPGRADE}; +use hyper::client::conn::http2; +use hyper::header::{CONNECTION, CONTENT_LENGTH, CONTENT_TYPE, HOST, HeaderValue, UPGRADE}; use hyper::service::service_fn; -use hyper::{Request, Response, StatusCode}; +use hyper::{Request, Response, StatusCode, Version}; use hyper_util::rt::{TokioExecutor, TokioIo}; use hyper_util::server::conn::auto; use rcgen::{BasicConstraints, CertificateParams, IsCa, KeyPair}; @@ -86,7 +89,11 @@ fn full(data: impl Into) -> TestBody { async fn route(req: Request) -> Result, Infallible> { Ok(match req.uri().path() { "/hello" => Response::new(full("hello")), - "/echo" => Response::new(full(req.into_body().collect().await.unwrap().to_bytes())), + // An upload cut short, as a cancelled one is, gets an empty answer. + "/echo" => match req.into_body().collect().await { + Ok(body) => Response::new(full(body.to_bytes())), + Err(_) => Response::builder().status(StatusCode::BAD_REQUEST).body(full("")).unwrap(), + }, "/authority" => Response::new(full(format!( "authority={} host={}", req.uri().authority().map_or("none".into(), |a| a.to_string()), @@ -97,10 +104,40 @@ async fn route(req: Request) -> Result, Infallible> // The stream ends with half the promised bytes, so the upstream resets it mid-body. "/cut" => late_end(10), "/ws" => websocket_echo(req), + "/trailers" => echo_with_trailers(req).await, + "/cookies" => { + let cookies = req.headers().get_all("cookie").iter(); + Response::new(full( + cookies.map(|c| c.to_str().unwrap()).collect::>().join(" | "), + )) + } + // Headers for this connection alone, which HTTP/2 forbids. + "/hop" => Response::builder() + .header(CONNECTION, "x-hop") + .header("x-hop", "1") + .header("keep-alive", "timeout=5") + .body(full("hop")) + .unwrap(), _ => Response::builder().status(StatusCode::NOT_FOUND).body(full("not found")).unwrap(), }) } +/// The request body back, with the request's trailers and one of the server's own. +async fn echo_with_trailers(req: Request) -> Response { + let collected = req.into_body().collect().await.unwrap(); + let mut trailers = collected.trailers().cloned().unwrap_or_default(); + trailers.insert("x-upstream-trailer", HeaderValue::from_static("yes")); + let frames = [ + Ok(Frame::data(collected.to_bytes())), + Ok(Frame::trailers(trailers)), + ]; + Response::builder() + // HTTP/1.1 sends only the trailers it names. + .header("trailer", "x-client-trailer, x-upstream-trailer") + .body(BodyExt::boxed(StreamBody::new(futures_util::stream::iter(frames)))) + .unwrap() +} + /// "hello" under a Content-Length of `content_length`, its end of stream sent a while after the /// last byte, as an HTTP/2 server does when its final DATA frame has no END_STREAM. fn late_end(content_length: usize) -> Response { @@ -139,15 +176,31 @@ fn websocket_echo(mut req: Request) -> Response { } async fn spawn_upstream(tls: Option>) -> SocketAddr { + spawn_upstream_with(tls, Arc::new(AtomicUsize::new(0)), route).await +} + +/// An upstream server answering with `handler`, counting the connections it accepts in +/// `accepts`. +async fn spawn_upstream_with( + tls: Option>, + accepts: Arc, + handler: F, +) -> SocketAddr +where + F: Fn(Request) -> Fut + Clone + Send + Sync + 'static, + Fut: Future, Infallible>> + Send + 'static, +{ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { loop { let (tcp, _) = listener.accept().await.unwrap(); + accepts.fetch_add(1, Ordering::SeqCst); let tls = tls.clone(); + let handler = handler.clone(); tokio::spawn(async move { let builder = auto::Builder::new(TokioExecutor::new()); - let svc = service_fn(route); + let svc = service_fn(handler); let _ = match tls { Some(config) => { let Ok(stream) = TlsAcceptor::from(config).accept(tcp).await else { @@ -193,8 +246,10 @@ impl Harness { format!("http://127.0.0.1:{}", self.proxy.port) } + /// An HTTP/1.1 client. The HTTP/2 tests below make their own. fn client_trusting(&self, root_pem: &str) -> reqwest::Client { reqwest::Client::builder() + .http1_only() .proxy(reqwest::Proxy::all(self.proxy_url()).unwrap()) .add_root_certificate(reqwest::Certificate::from_pem(root_pem.as_bytes()).unwrap()) .build() @@ -261,8 +316,7 @@ impl Harness { loop { let event = self.events.recv_timeout(Duration::from_secs(10)).expect("flow did not end"); - let done = - matches!(event, ProxyEvent::ResponseBodyComplete { .. } | ProxyEvent::Error { .. }); + let done = is_terminal(&event); events.push(event); if done { break; @@ -273,6 +327,87 @@ impl Harness { } events } + + /// Events of the next `n` flows, which may interleave, by flow id. + fn flows_events(&self, n: usize) -> HashMap> { + let mut flows: HashMap> = HashMap::new(); + let mut ended = 0; + while ended < n { + let event = + self.events.recv_timeout(Duration::from_secs(10)).expect("flows did not end"); + ended += usize::from(is_terminal(&event)); + flows.entry(flow_id(&event)).or_default().push(event); + } + while let Ok(event) = self.events.recv_timeout(Duration::from_millis(100)) { + flows.entry(flow_id(&event)).or_default().push(event); + } + flows + } + + /// TLS to `localhost:port` through a CONNECT tunnel, trusting the proxy's CA and offering + /// `alpn`. + async fn tls_tunnel( + &self, + port: u16, + alpn: &[&[u8]], + ) -> tokio_rustls::client::TlsStream { + let mut config = ClientConfig::clone(&self.tls_client_config()); + config.alpn_protocols = alpn.iter().map(|p| p.to_vec()).collect(); + let tcp = self.connect_tunnel(port).await; + let server_name = ServerName::try_from("localhost").unwrap(); + TlsConnector::from(Arc::new(config)).connect(server_name, tcp).await.unwrap() + } + + /// An HTTP/2 connection to `localhost:port` through the proxy, offering both versions as a + /// browser does. + async fn h2_client(&self, port: u16) -> http2::SendRequest { + let tls = self.tls_tunnel(port, &[b"h2", b"http/1.1"]).await; + assert_eq!(tls.get_ref().1.alpn_protocol(), Some(&b"h2"[..])); + let (sender, conn) = + http2::handshake(TokioExecutor::new(), TokioIo::new(tls)).await.unwrap(); + tokio::spawn(conn); + sender + } +} + +fn is_terminal(event: &ProxyEvent) -> bool { + matches!( + event, + ProxyEvent::ResponseBodyComplete { .. } + | ProxyEvent::Error { .. } + | ProxyEvent::Cancelled { .. } + ) +} + +fn flow_id(event: &ProxyEvent) -> u64 { + match event { + ProxyEvent::RequestStart { id, .. } + | ProxyEvent::RequestHeader { id, .. } + | ProxyEvent::RequestBodyChunk { id, .. } + | ProxyEvent::RequestBodyComplete { id, .. } + | ProxyEvent::ResponseStart { id, .. } + | ProxyEvent::ResponseHeader { id, .. } + | ProxyEvent::ResponseBodyChunk { id, .. } + | ProxyEvent::ResponseBodyComplete { id, .. } + | ProxyEvent::Error { id, .. } + | ProxyEvent::Cancelled { id, .. } => *id, + } +} + +/// The headers a flow captured on one side, in order. +fn headers(events: &[ProxyEvent], request: bool) -> Vec<(String, String)> { + events + .iter() + .filter_map(|e| match e { + ProxyEvent::RequestHeader { name, value, .. } if request => { + Some((name.clone(), value.clone())) + } + ProxyEvent::ResponseHeader { name, value, .. } if !request => { + Some((name.clone(), value.clone())) + } + _ => None, + }) + .collect() } /// The captured bytes of one side of a flow, and how many bytes passed through. @@ -797,3 +932,460 @@ async fn plain_http_request_looped_back_to_the_proxy_stops() { let error = find!(events, ProxyEvent::Error { error, .. } => error.clone()); assert_eq!(error, "request looped back to Yaak Proxy"); } + +// -- HTTP/2 clients -- + +fn h2_request(method: &str, port: u16, path: &str, body: TestBody) -> Request { + Request::builder() + .method(method) + .uri(format!("https://localhost:{port}{path}")) + .body(body) + .unwrap() +} + +/// The upstream test server, speaking HTTP/1.1 alone. +fn h1_only(server: &ServerConfig) -> Arc { + let mut server = server.clone(); + server.alpn_protocols = vec![b"http/1.1".to_vec()]; + Arc::new(server) +} + +#[tokio::test(flavor = "multi_thread")] +async fn h2_client_to_an_h2_upstream() { + let h = Harness::start(1024); + let upstream = spawn_upstream(Some(h.pki.server.clone())).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + + let resp = client.send_request(h2_request("GET", port, "/authority", full(""))).await.unwrap(); + assert_eq!((resp.status(), resp.version()), (StatusCode::OK, Version::HTTP_2)); + let text = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(text, format!("authority=localhost:{port} host=none")); + + let events = h.flow_events(); + let (method, url, version, connection) = find!(events, + ProxyEvent::RequestStart { method, url, http_version, connection, .. } => + (method.clone(), url.clone(), http_version.clone(), *connection)); + assert_eq!(method, "GET"); + assert_eq!(url, format!("https://localhost:{port}/authority")); + assert_eq!(version, "HTTP/2"); + assert_eq!(connection, 1); + // `:authority` is shown as `Host`, once and first, as an HTTP/1.1 request has it. + let request_headers = headers(&events, true); + assert_eq!(request_headers.first(), Some(&("host".to_string(), format!("localhost:{port}")))); + assert_eq!(request_headers.iter().filter(|(name, _)| name == "host").count(), 1); + + let (status, version, remote_addr) = find!(events, + ProxyEvent::ResponseStart { status, http_version, remote_addr, .. } => + (*status, http_version.clone(), *remote_addr)); + assert_eq!((status, version.as_str(), remote_addr), (200, "HTTP/2", Some(upstream))); + let (size, error, cancelled) = find!(events, + ProxyEvent::ResponseBodyComplete { size, error, cancelled, .. } => + (*size, error.clone(), *cancelled)); + assert_eq!((size, error, cancelled), (text.len() as u64, None, false)); + assert_eq!(body(&events, false).0, text); +} + +#[tokio::test(flavor = "multi_thread")] +async fn h2_client_to_an_h1_upstream() { + let h = Harness::start(1024); + let upstream = spawn_upstream(Some(h1_only(&h.pki.server))).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + + let resp = client.send_request(h2_request("POST", port, "/echo", full("name=yaak"))).await; + let resp = resp.unwrap(); + assert_eq!((resp.status(), resp.version()), (StatusCode::OK, Version::HTTP_2)); + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "name=yaak"); + let events = h.flow_events(); + let version = find!(events, ProxyEvent::RequestStart { http_version, .. } => http_version); + assert_eq!(version, "HTTP/2"); + let (version, remote_addr) = find!(events, + ProxyEvent::ResponseStart { http_version, remote_addr, .. } => + (http_version.clone(), *remote_addr)); + assert_eq!((version.as_str(), remote_addr), ("HTTP/1.1", Some(upstream))); + assert_eq!(body(&events, true), (b"name=yaak".to_vec(), 9)); + + // `:authority` goes upstream as `Host`. + let resp = client.send_request(h2_request("GET", port, "/authority", full(""))).await; + let text = resp.unwrap().into_body().collect().await.unwrap().to_bytes(); + assert_eq!(text, format!("authority=none host=localhost:{port}")); + h.flow_events(); + + // Headers for the HTTP/1.1 connection alone are captured but not sent on to the client. + let resp = client.send_request(h2_request("GET", port, "/hop", full(""))).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + for name in ["connection", "x-hop", "keep-alive"] { + assert!(!resp.headers().contains_key(name), "{name} in {:?}", resp.headers()); + } + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "hop"); + let captured = headers(&h.flow_events(), false); + assert!(captured.contains(&("x-hop".to_string(), "1".to_string())), "{captured:?}"); +} + +/// Each stream is a flow of its own, all on the client's one connection, and they share one +/// connection upstream when the server speaks HTTP/2. +#[tokio::test(flavor = "multi_thread")] +async fn concurrent_h2_streams_are_flows_of_their_own() { + let h = Harness::start(1024); + let accepts = Arc::new(AtomicUsize::new(0)); + let upstream = spawn_upstream_with(Some(h.pki.server.clone()), accepts.clone(), route).await; + let port = upstream.port(); + let client = h.h2_client(port).await; + + const STREAMS: usize = 20; + let sent = (0..STREAMS).map(|i| { + let mut client = client.clone(); + async move { + let req = h2_request("POST", port, "/echo", full(format!("stream {i}"))); + let resp = client.send_request(req).await.unwrap(); + resp.into_body().collect().await.unwrap().to_bytes() + } + }); + let bodies = futures_util::future::join_all(sent).await; + for (i, body) in bodies.iter().enumerate() { + assert_eq!(body, format!("stream {i}").as_bytes()); + } + + let flows = h.flows_events(STREAMS); + assert_eq!(flows.len(), STREAMS); + let connections: HashSet = flows + .values() + .map(|events| find!(events, ProxyEvent::RequestStart { connection, .. } => *connection)) + .collect(); + assert_eq!(connections.len(), 1, "one client connection"); + let mut captured: Vec = + flows.values().map(|events| String::from_utf8(body(events, true).0).unwrap()).collect(); + captured.sort(); + let mut expected: Vec = (0..STREAMS).map(|i| format!("stream {i}")).collect(); + expected.sort(); + assert_eq!(captured, expected); + for events in flows.values() { + let echoed = body(events, false).0; + assert_eq!(echoed, body(events, true).0, "each flow keeps its own response"); + } + assert_eq!(accepts.load(Ordering::SeqCst), 1, "the streams share one upstream connection"); +} + +/// An HTTP/1.1 server takes the streams on connections of its own, at once rather than one +/// after another, and they are reused after. +#[tokio::test(flavor = "multi_thread")] +async fn concurrent_h2_streams_to_an_h1_upstream_run_at_once() { + const STREAMS: usize = 10; + let h = Harness::start(1024); + let accepts = Arc::new(AtomicUsize::new(0)); + // Each `/together` request is held until all of them are in flight at once, which they can + // only be if none waits for another. Counted, not timed: how long a connection takes to + // open varies by machine. + let in_flight = Arc::new(AtomicUsize::new(0)); + let most = Arc::new(AtomicUsize::new(0)); + let handler = { + let (in_flight, most) = (in_flight.clone(), most.clone()); + move |req: Request| { + let (in_flight, most) = (in_flight.clone(), most.clone()); + async move { + if req.uri().path() != "/together" { + return route(req).await; + } + most.fetch_max(in_flight.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst); + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + while in_flight.load(Ordering::SeqCst) < STREAMS + && tokio::time::Instant::now() < deadline + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + Ok(Response::new(full("together"))) + } + } + }; + let upstream = + spawn_upstream_with(Some(h1_only(&h.pki.server)), accepts.clone(), handler).await; + let port = upstream.port(); + let client = h.h2_client(port).await; + + let sent = (0..STREAMS).map(|_| { + let mut client = client.clone(); + async move { + let resp = client.send_request(h2_request("GET", port, "/together", full(""))).await; + resp.unwrap().into_body().collect().await.unwrap().to_bytes() + } + }); + for body in futures_util::future::join_all(sent).await { + assert_eq!(body, "together"); + } + assert_eq!(most.load(Ordering::SeqCst), STREAMS, "streams in flight upstream at once"); + assert_eq!(h.flows_events(STREAMS).len(), STREAMS); + let opened = accepts.load(Ordering::SeqCst); + assert!((2..=STREAMS).contains(&opened), "{opened} upstream connections"); + + let mut client = client.clone(); + let resp = client.send_request(h2_request("GET", port, "/hello", full(""))).await.unwrap(); + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "hello"); + h.flow_events(); + assert_eq!(accepts.load(Ordering::SeqCst), opened, "an idle connection is reused"); +} + +/// Sends its message when dropped, so a test sees the server give up on a request. +struct OnDrop(tokio::sync::mpsc::UnboundedSender<&'static str>, &'static str); + +impl Drop for OnDrop { + fn drop(&mut self) { + let _ = self.0.send(self.1); + } +} + +/// An upstream whose `/hang` never answers and whose `/stall` sends "hello" and then nothing, +/// reporting on `tx` when each request arrives and when it is dropped. +async fn spawn_stalling_upstream( + tls: Arc, + tx: tokio::sync::mpsc::UnboundedSender<&'static str>, +) -> SocketAddr { + let handler = move |req: Request| { + let tx = tx.clone(); + async move { + match req.uri().path() { + "/hang" => { + let _guard = OnDrop(tx.clone(), "hang dropped"); + let _ = tx.send("hang arrived"); + match std::future::pending::().await {} + } + _ => { + let guard = OnDrop(tx, "stall dropped"); + // The second frame never comes, and the guard goes with the body. + let frames = + futures_util::stream::unfold((true, guard), |(first, guard)| async move { + if first { + let _ = guard.0.send("stall sent"); + let hello = Ok(Frame::data(Bytes::from("hello"))); + return Some((hello, (false, guard))); + } + let _guard = guard; + std::future::pending().await + }); + Ok(Response::new(BodyExt::boxed(StreamBody::new(frames)))) + } + } + } + }; + spawn_upstream_with(Some(tls), Arc::new(AtomicUsize::new(0)), handler).await +} + +async fn next_message(rx: &mut tokio::sync::mpsc::UnboundedReceiver<&'static str>) -> &'static str { + tokio::time::timeout(Duration::from_secs(10), rx.recv()).await.unwrap().unwrap() +} + +/// A stream the client resets before the response is a cancelled flow, not an error, and the +/// request upstream is cancelled with it. +#[tokio::test(flavor = "multi_thread")] +async fn h2_stream_cancelled_before_the_response() { + let h = Harness::start(1024); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let upstream = spawn_stalling_upstream(h.pki.server.clone(), tx).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + + let pending = tokio::spawn({ + let mut client = client.clone(); + async move { client.send_request(h2_request("GET", port, "/hang", full(""))).await } + }); + assert_eq!(next_message(&mut rx).await, "hang arrived"); + pending.abort(); + assert_eq!(next_message(&mut rx).await, "hang dropped"); + + let events = h.flow_events(); + find!(events, ProxyEvent::Cancelled { .. } => ()); + assert!(!events.iter().any(|e| matches!(e, ProxyEvent::Error { .. })), "{events:#?}"); + + // The connection carries on. + let resp = client.send_request(h2_request("GET", port, "/stall", full(""))).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(next_message(&mut rx).await, "stall sent"); +} + +/// A stream the client resets partway through the response body is cancelled too. +#[tokio::test(flavor = "multi_thread")] +async fn h2_stream_cancelled_during_the_body() { + let h = Harness::start(1024); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let upstream = spawn_stalling_upstream(h.pki.server.clone(), tx).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + + let resp = client.send_request(h2_request("GET", port, "/stall", full(""))).await.unwrap(); + let mut body = resp.into_body(); + let hello = body.frame().await.unwrap().unwrap().into_data().unwrap(); + assert_eq!(hello, "hello"); + assert_eq!(next_message(&mut rx).await, "stall sent"); + drop(body); + assert_eq!(next_message(&mut rx).await, "stall dropped"); + + let events = h.flow_events(); + let (size, error, cancelled) = find!(events, + ProxyEvent::ResponseBodyComplete { size, error, cancelled, .. } => + (*size, error.clone(), *cancelled)); + assert_eq!((size, error, cancelled), (5, None, true)); +} + +/// Trailers go both ways, as gRPC needs them to. +#[tokio::test(flavor = "multi_thread")] +async fn h2_trailers_pass_through() { + let h = Harness::start(1024); + for tls in [h.pki.server.clone(), h1_only(&h.pki.server)] { + let upstream = spawn_upstream(Some(tls)).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + + let mut trailers = hyper::HeaderMap::new(); + trailers.insert("x-client-trailer", HeaderValue::from_static("1")); + let frames = [ + Ok(Frame::data(Bytes::from("message"))), + Ok(Frame::trailers(trailers)), + ]; + let upload = BodyExt::boxed(StreamBody::new(futures_util::stream::iter(frames))); + let mut req = h2_request("POST", port, "/trailers", upload); + req.headers_mut().insert("te", HeaderValue::from_static("trailers")); + // HTTP/1.1 carries only trailers named up front, so an HTTP/1.1 server gets this one. + req.headers_mut().insert("trailer", HeaderValue::from_static("x-client-trailer")); + + let resp = client.send_request(req).await.unwrap(); + let collected = resp.into_body().collect().await.unwrap(); + let trailers = collected.trailers().cloned().unwrap(); + assert_eq!(trailers["x-client-trailer"], "1"); + assert_eq!(trailers["x-upstream-trailer"], "yes"); + assert_eq!(collected.to_bytes(), "message"); + + let events = h.flow_events(); + let (size, error) = find!(events, + ProxyEvent::ResponseBodyComplete { size, error, .. } => (*size, error.clone())); + assert_eq!((size, error), (7, None)); + assert_eq!(body(&events, true), (b"message".to_vec(), 7)); + } +} + +/// HTTP/2 may send a field per cookie; an HTTP/1.1 server gets them as the one `Cookie` it +/// expects. +#[tokio::test(flavor = "multi_thread")] +async fn h2_cookies_are_joined_for_an_h1_upstream() { + let h = Harness::start(1024); + for (tls, expected) in [ + (h1_only(&h.pki.server), "theme=dark; session=abc"), + (h.pki.server.clone(), "theme=dark | session=abc"), + ] { + let upstream = spawn_upstream(Some(tls)).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + let mut req = h2_request("GET", port, "/cookies", full("")); + req.headers_mut().append("cookie", HeaderValue::from_static("theme=dark")); + req.headers_mut().append("cookie", HeaderValue::from_static("session=abc")); + let resp = client.send_request(req).await.unwrap(); + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), expected); + h.flow_events(); + } +} + +/// A client that resets its stream partway through an upload cancels the flow, whichever +/// version the server speaks. +#[tokio::test(flavor = "multi_thread")] +async fn h2_upload_cancelled_by_the_client() { + let h = Harness::start(1024); + for tls in [h1_only(&h.pki.server), h.pki.server.clone()] { + let upstream = spawn_upstream(Some(tls)).await; + let port = upstream.port(); + let client = h.h2_client(port).await; + + let (tx, rx) = tokio::sync::mpsc::unbounded_channel::, Infallible>>(); + tx.send(Ok(Frame::data(Bytes::from("first part")))).unwrap(); + let frames = futures_util::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|frame| (frame, rx)) + }); + let upload = BodyExt::boxed(StreamBody::new(frames)); + let pending = tokio::spawn({ + let mut client = client.clone(); + async move { client.send_request(h2_request("POST", port, "/echo", upload)).await } + }); + tokio::time::sleep(Duration::from_millis(300)).await; + pending.abort(); + drop(tx); + + let events = h.flow_events(); + find!(events, ProxyEvent::Cancelled { .. } => ()); + assert!(!events.iter().any(|e| matches!(e, ProxyEvent::Error { .. })), "{events:#?}"); + } +} + +/// An HTTP/2 request's body has no length of its own, which HTTP/1.1 needs, even on a GET. +#[tokio::test(flavor = "multi_thread")] +async fn h2_streamed_body_reaches_an_h1_upstream() { + let h = Harness::start(1024); + let upstream = spawn_upstream(Some(h1_only(&h.pki.server))).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + + for method in ["GET", "POST"] { + let frames = ["streamed ", "body"].map(|s| Ok(Frame::data(Bytes::from(s)))); + let upload = BodyExt::boxed(StreamBody::new(futures_util::stream::iter(frames))); + let resp = client.send_request(h2_request(method, port, "/echo", upload)).await.unwrap(); + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "streamed body"); + let events = h.flow_events(); + assert_eq!(body(&events, true), (b"streamed body".to_vec(), 13), "{method}"); + } +} + +/// More than the flow-control windows hold, both ways, through an HTTP/2 and an HTTP/1.1 server. +#[tokio::test(flavor = "multi_thread")] +async fn h2_large_bodies_pass_flow_control() { + const LARGE: usize = 12 * 1024 * 1024; + let h = Harness::start(1024); + for tls in [h.pki.server.clone(), h1_only(&h.pki.server)] { + let upstream = spawn_upstream(Some(tls)).await; + let port = upstream.port(); + let mut client = h.h2_client(port).await; + + let chunk = Bytes::from(vec![b'y'; BIG_BODY]); + let chunks = (0..LARGE / BIG_BODY).map(move |_| Ok(Frame::data(chunk.clone()))); + let upload = BodyExt::boxed(StreamBody::new(futures_util::stream::iter(chunks))); + let resp = client.send_request(h2_request("POST", port, "/echo", upload)).await.unwrap(); + let echoed = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(echoed.len(), LARGE); + assert!(echoed.iter().all(|b| *b == b'y')); + + let events = h.flow_events(); + let (size, truncated) = find!(events, + ProxyEvent::RequestBodyComplete { size, truncated, .. } => (*size, *truncated)); + assert_eq!((size, truncated), (LARGE as u64, true)); + let (size, truncated, error) = find!(events, + ProxyEvent::ResponseBodyComplete { size, truncated, error, .. } => + (*size, *truncated, error.clone())); + assert_eq!((size, truncated, error), (LARGE as u64, true, None)); + assert_eq!(body(&events, false), (vec![b'y'; 1024], LARGE)); + } +} + +/// With HTTP/2 off, a client that offers both gets HTTP/1.1, from the next connection on. +#[tokio::test(flavor = "multi_thread")] +async fn http1_only_makes_h2_clients_fall_back() { + let h = Harness::start(1024); + let upstream = spawn_upstream(Some(h.pki.server.clone())).await; + let port = upstream.port(); + + h.proxy.set_http2(false); + let tls = h.tls_tunnel(port, &[b"h2", b"http/1.1"]).await; + assert_eq!(tls.get_ref().1.alpn_protocol(), Some(&b"http/1.1"[..])); + let (mut sender, conn) = + hyper::client::conn::http1::handshake(TokioIo::new(tls)).await.unwrap(); + tokio::spawn(conn); + let req = Request::get("/hello").header(HOST, format!("localhost:{port}")).body(full("")); + let resp = sender.send_request(req.unwrap()).await.unwrap(); + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "hello"); + let events = h.flow_events(); + let version = find!(events, ProxyEvent::RequestStart { http_version, .. } => http_version); + assert_eq!(version, "HTTP/1.1"); + + h.proxy.set_http2(true); + let mut client = h.h2_client(port).await; + let resp = client.send_request(h2_request("GET", port, "/hello", full(""))).await.unwrap(); + assert_eq!(resp.version(), Version::HTTP_2); + let events = h.flow_events(); + let version = find!(events, ProxyEvent::RequestStart { http_version, .. } => http_version); + assert_eq!(version, "HTTP/2"); +}