leyline/core/session/execute/
response.rs1use std::sync::Arc;
2use std::time::SystemTime;
3
4use url::Url;
5
6use super::super::decompress::{decompress_and_strip, drain_stream_into_vec};
7use super::journey::Journey;
8use crate::core::Session;
9use crate::core::deadline::Deadline;
10use crate::core::device::{SessionState, StateParts, unix_secs};
11use crate::core::error::Result;
12use crate::core::response::Response;
13use crate::core::transport::{ResponseMode, TransportResponse};
14use crate::util::lock;
15
16impl Session {
17 pub(super) fn store_cookies(
18 &self,
19 resp_headers: &[(http::HeaderName, http::HeaderValue)],
20 url: &Url,
21 ) {
22 let set_cookies: Vec<&str> = resp_headers
23 .iter()
24 .filter(|(k, _)| *k == "set-cookie")
25 .filter_map(|(_, v)| v.to_str().ok())
26 .collect();
27 if !set_cookies.is_empty() {
28 self.inner
29 .cookie_jar
30 .store_response_cookies(set_cookies.as_slice(), url);
31 }
32 }
33
34 #[must_use]
35 pub fn state(&self) -> SessionState {
36 let now = SystemTime::now();
37 SessionState::from_parts(StateParts {
38 tls_sessions: self.inner.connector.session_cache().export(unix_secs(now)),
39 #[cfg(feature = "http3")]
40 alt_svc: self.inner.pool.alt_svc_entries(),
41 #[cfg(not(feature = "http3"))]
42 alt_svc: Vec::new(),
43 hsts: lock(&self.inner.pool.hsts).export(now),
44 })
45 }
46
47 pub(crate) fn restore_state(&self, parts: &StateParts) {
48 let now = SystemTime::now();
49 self.inner
50 .connector
51 .session_cache()
52 .import(&parts.tls_sessions, unix_secs(now));
53 #[cfg(feature = "http3")]
54 self.inner.pool.restore_alt_svc(&parts.alt_svc);
55 lock(&self.inner.pool.hsts).import(&parts.hsts, now);
56 }
57
58 pub(super) fn note_hsts(
59 &self,
60 url: &Url,
61 resp_headers: &[(http::HeaderName, http::HeaderValue)],
62 ) {
63 if self.inner.tls_trust.accepts_invalid_certs() {
64 return;
65 }
66 let values: Vec<&str> = resp_headers
67 .iter()
68 .filter(|(k, _)| *k == "strict-transport-security")
69 .filter_map(|(_, v)| v.to_str().ok())
70 .collect();
71 if !values.is_empty() {
72 lock(&self.inner.pool.hsts).note(url, &values, SystemTime::now());
73 }
74 }
75
76 pub(super) fn hsts_upgrade(&self, url: &Arc<Url>) -> Arc<Url> {
77 lock(&self.inner.pool.hsts)
78 .upgrade(url, SystemTime::now())
79 .map_or_else(|| Arc::clone(url), Arc::new)
80 }
81
82 #[cfg(feature = "http3")]
83 pub(super) fn note_alt_svc(
84 &self,
85 url: &Url,
86 resp_headers: &[(http::HeaderName, http::HeaderValue)],
87 ) {
88 let owned: Vec<String> = resp_headers
89 .iter()
90 .filter(|(k, _)| *k == "alt-svc")
91 .map(|(_, v)| String::from_utf8_lossy(v.as_bytes()).into_owned())
92 .collect();
93 let fields: Vec<&str> = owned.iter().map(String::as_str).collect();
94 if fields.is_empty() {
95 return;
96 }
97 let age = resp_headers
98 .iter()
99 .find(|(k, _)| *k == "age")
100 .and_then(|(_, v)| v.to_str().ok())
101 .and_then(|v| crate::util::delta_seconds(v.split(',').next().unwrap_or_default()))
102 .map_or(std::time::Duration::ZERO, std::time::Duration::from_secs);
103 if let Some(host) = url.host_str()
104 && let Some(port) = url.port_or_known_default()
105 {
106 self.inner.pool.note_alt_svc(host, port, &fields, age);
107 }
108 }
109
110 pub(super) async fn assemble_response(
111 &self,
112 leg: TransportResponse,
113 journey: Journey,
114 audit_headers: Vec<(String, String)>,
115 response: ResponseMode,
116 deadline: &Deadline,
117 ) -> Result<Response> {
118 let TransportResponse {
119 status,
120 headers,
121 trailers,
122 body,
123 final_url,
124 version,
125 tls,
126 ..
127 } = leg;
128 let (body, headers) = self
129 .finalize_response_body(
130 body,
131 headers,
132 response.keeps_stream(status.as_u16()),
133 deadline,
134 )
135 .await?;
136 let audited = self.inner.audit_tls.is_some();
137 let mut audit_headers = audit_headers;
138 audit_headers.retain(|(name, value)| crate::core::transport::sent_on(version, name, value));
139 Ok(Response {
140 status,
141 headers: headers.into_iter().collect(),
142 body,
143 url: final_url,
144 redirect_chain: journey.chain,
145 version,
146 trailers,
147 request_headers: audit_headers,
148 tls,
149 request_method: if audited {
150 journey.method
151 } else {
152 String::new()
153 },
154 audit_tls: self.inner.audit_tls.clone(),
155 audit_cache: std::sync::OnceLock::new(),
156 compression: self.inner.compression,
157 timing: journey.timing,
158 attempts: 1,
159 proxy: None,
160 })
161 }
162
163 pub(super) async fn finalize_response_body(
164 &self,
165 resp_body_shape: crate::core::transport::TransportBody,
166 resp_headers: Vec<(http::HeaderName, http::HeaderValue)>,
167 keep_stream: bool,
168 deadline: &Deadline,
169 ) -> Result<(
170 crate::core::response::ResponseBody,
171 Vec<(http::HeaderName, http::HeaderValue)>,
172 )> {
173 Ok(match resp_body_shape {
174 crate::core::transport::TransportBody::Streaming(mut bs) if keep_stream => {
175 bs.set_read_timeout(deadline.read());
176 bs.set_body_timeout(deadline.body());
177 bs.stop_on(&self.inner.shutdown);
178 bs.watch_end();
179 (
180 crate::core::response::ResponseBody::Streaming(bs),
181 resp_headers,
182 )
183 }
184 crate::core::transport::TransportBody::Streaming(mut bs) => {
185 bs.set_read_timeout(deadline.read());
186 bs.set_body_timeout(deadline.body());
187 let buf = drain_stream_into_vec(bs, self.inner.compression.max_body_size).await?;
188 let (buf, resp_headers) =
189 decompress_and_strip(buf, resp_headers, &self.inner.compression)?;
190 (
191 crate::core::response::ResponseBody::Buffered(buf),
192 resp_headers,
193 )
194 }
195 crate::core::transport::TransportBody::Buffered(buf) => {
196 if keep_stream {
197 let mut bs =
198 crate::core::body_stream::BodyStream::from_bytes(bytes::Bytes::from(buf));
199 bs.stop_on(&self.inner.shutdown);
200 bs.watch_end();
201 (
202 crate::core::response::ResponseBody::Streaming(bs),
203 resp_headers,
204 )
205 } else {
206 let (buf, resp_headers) =
207 decompress_and_strip(buf, resp_headers, &self.inner.compression)?;
208 (
209 crate::core::response::ResponseBody::Buffered(buf),
210 resp_headers,
211 )
212 }
213 }
214 })
215 }
216}