Skip to main content

leyline/core/session/
execute.rs

1use url::Url;
2
3use crate::FetchSite;
4use crate::core::transport::Prepared;
5use std::sync::Arc;
6
7use crate::profile::Preset;
8
9use self::auth::{DigestLeg, unreplayable_body};
10use self::journey::{Journey, redirect_location};
11use super::Session;
12use crate::core::body::Body;
13use crate::core::config::TimeoutConfig;
14use crate::core::deadline::Deadline;
15use crate::core::digest::DigestAuth;
16use crate::core::error::{Error, Kind, Result};
17use crate::core::headers::HeaderList;
18use crate::core::response::{Response, ResponseBody};
19use crate::core::{ProxyConfig, RedirectPolicy};
20use crate::trace;
21use crate::util::{lock, redact, without_userinfo};
22
23mod auth;
24mod headers;
25mod journey;
26mod response;
27
28pub(crate) struct Attempt {
29    pub(crate) method: http::Method,
30    pub(crate) url: String,
31    pub(crate) preset: Option<Preset>,
32    pub(crate) body: Body,
33    pub(crate) headers: Option<HeaderList>,
34    pub(crate) deadline: Deadline,
35    pub(crate) stream_response: bool,
36    pub(crate) proxy: Option<ProxyConfig>,
37    pub(crate) header_order: Option<Vec<String>>,
38    pub(crate) redirect: Option<RedirectPolicy>,
39    pub(crate) digest: Option<DigestAuth>,
40    pub(crate) initiator: Option<Url>,
41    pub(crate) trusted_origin: Option<Url>,
42}
43
44impl Attempt {
45    pub(crate) fn again(&self, body: Body, headers: Option<HeaderList>) -> Attempt {
46        Attempt {
47            method: self.method.clone(),
48            url: self.url.clone(),
49            preset: self.preset,
50            body,
51            headers,
52            deadline: self.deadline,
53            stream_response: self.stream_response,
54            proxy: self.proxy.clone(),
55            header_order: self.header_order.clone(),
56            redirect: self.redirect.clone(),
57            digest: self.digest.clone(),
58            initiator: self.initiator.clone(),
59            trusted_origin: self.trusted_origin.clone(),
60        }
61    }
62}
63
64impl Session {
65    pub(crate) fn deadline(&self, request: Option<&TimeoutConfig>) -> Deadline {
66        Deadline::new(&self.inner.timeouts, request)
67    }
68
69    pub(crate) async fn attempt(&self, attempt: Attempt) -> Result<Response> {
70        let deadline = attempt.deadline;
71        let inner: std::pin::Pin<
72            Box<dyn std::future::Future<Output = Result<Response>> + Send + '_>,
73        > = Box::pin(self.execute_inner(attempt));
74        trace::scope(self.inner.trace.as_ref(), async move {
75            let out = self.unless_shut_down(deadline.total(inner)).await;
76            trace::done(match &out {
77                Ok(_) => Ok(()),
78                Err(e) => Err(e),
79            });
80            out
81        })
82        .await
83    }
84
85    pub(crate) async fn unless_shut_down<T>(
86        &self,
87        fut: impl std::future::Future<Output = Result<T>>,
88    ) -> Result<T> {
89        tokio::select! {
90            biased;
91            () = self.inner.shutdown.cancelled() => Err(Error::shut_down()),
92            out = fut => out,
93        }
94    }
95
96    pub fn shutdown(&self) {
97        self.inner.shutdown.cancel();
98    }
99
100    pub fn is_shut_down(&self) -> bool {
101        self.inner.shutdown.is_cancelled()
102    }
103
104    pub(crate) async fn traced<F>(
105        &self,
106        method: &str,
107        original_url: &str,
108        streamed: bool,
109        request: F,
110    ) -> Result<Response>
111    where
112        F: std::future::Future<Output = (Result<Response>, u32)>,
113    {
114        let hook = self.inner.trace.as_ref();
115        trace::scope(hook, async move {
116            let (out, attempts) = request.await;
117            if hook.is_some() {
118                let finish = trace::Finish {
119                    method,
120                    original_url,
121                    attempts,
122                    streamed,
123                };
124                trace::summary(&finish, out.as_ref());
125            }
126            out
127        })
128        .await
129    }
130
131    #[tracing::instrument(
132        name = "session.execute",
133        level = "debug",
134        skip_all,
135        fields(http.method = attempt.method.as_str(), http.url = redact(&attempt.url))
136    )]
137    async fn execute_inner(&self, attempt: Attempt) -> Result<Response> {
138        let Attempt {
139            method,
140            url: raw_url,
141            preset,
142            body,
143            headers: extra_headers,
144            deadline,
145            stream_response,
146            proxy: request_proxy,
147            header_order,
148            redirect,
149            digest,
150            initiator,
151            trusted_origin,
152        } = attempt;
153        let redirect_policy = redirect.as_ref().unwrap_or(&self.inner.redirect_policy);
154        let request_proxy = request_proxy.as_ref();
155        let header_order = header_order.as_deref();
156        let mut journey = Journey::begin(
157            self.hsts_upgrade(&self.resolve_url(&raw_url)?),
158            method.to_string(),
159            body,
160            extra_headers,
161            initiator,
162            self.session_referer(),
163        )
164        .trusting(trusted_origin.as_ref());
165
166        let mut digest = digest.map(DigestLeg::new);
167        let redirect_cap = redirect_policy.max_redirects_hint();
168        let mut pass = None;
169        loop {
170            journey.url = self.hsts_upgrade(&journey.url);
171            drop(pass.take());
172            pass = self.inner.host_limits.admit(&journey.url).await;
173            let headers = self.leg_headers(&journey, preset, header_order);
174            journey.authorized = None;
175            let audit_headers = self.audit_copy(&headers);
176
177            let step_body = std::mem::take(&mut journey.body);
178            let replay_body = step_body.replay();
179
180            let proxy = self.proxy_for(&journey.url, request_proxy)?;
181            let send = self.send_with_policy(Prepared {
182                method: &journey.method,
183                url: &journey.url,
184                headers,
185                body: step_body,
186                proxy,
187                stream_response,
188            });
189            let leg = deadline.response_header(send).await?;
190            journey.timing.add_leg(&leg.timing);
191
192            self.store_cookies(&leg.headers, &journey.url);
193            self.note_hsts(&journey.url, &leg.headers);
194            #[cfg(feature = "http3")]
195            self.note_alt_svc(&journey.url, &leg.headers);
196
197            let next = Turn {
198                code: leg.status.as_u16(),
199                headers: &leg.headers,
200                redirect_policy,
201                redirect_cap,
202            };
203            if next.take(&mut journey, digest.as_mut(), replay_body)? {
204                drop(leg.body);
205                continue;
206            }
207
208            let mut response = self
209                .assemble_response(leg, journey, audit_headers, stream_response, &deadline)
210                .await?;
211            if let ResponseBody::Streaming(body) = &mut response.body {
212                body.hold(pass.take());
213            }
214            return Ok(response);
215        }
216    }
217
218    fn session_referer(&self) -> Option<&str> {
219        self.inner
220            .default_headers
221            .iter()
222            .rfind(|(k, _)| k.eq_ignore_ascii_case("referer"))
223            .map(|(_, v)| v.as_str())
224            .filter(|v| !v.is_empty())
225    }
226
227    fn resolve_url(&self, raw_url: &str) -> Result<Arc<Url>> {
228        let mut cache = lock(&self.inner.url_cache);
229        match cache.as_mut() {
230            Some((raw, parsed)) if raw == raw_url => Ok(Arc::clone(parsed)),
231            _ => {
232                let parsed = Arc::new(Url::parse(raw_url).map_err(Error::from_url_parse)?);
233                *cache = Some((raw_url.to_string(), Arc::clone(&parsed)));
234                Ok(parsed)
235            }
236        }
237    }
238}
239
240pub(super) struct RequestContext<'a> {
241    pub(super) origin: &'a str,
242    pub(super) origin_downgrade: bool,
243    pub(super) referer: &'a str,
244    pub(super) initiated: bool,
245    pub(super) site: SiteContext<'a>,
246}
247
248pub(in crate::core::session) struct SiteContext<'a> {
249    pub(in crate::core::session) fetch_site: FetchSite,
250    pub(in crate::core::session) initiator: Option<&'a Url>,
251    pub(in crate::core::session) chain: &'a [Url],
252}
253
254fn fetch_site_for(initiator: Option<&Url>, chain: &[Url], current: &Url) -> FetchSite {
255    let Some(initiator) = initiator else {
256        return FetchSite::CrossSite;
257    };
258    FetchSite::across(initiator, chain.iter().chain(std::iter::once(current)))
259}
260
261pub(crate) fn url_origin(url: &Url) -> String {
262    let host = url.host_str().unwrap_or("");
263    match url.port() {
264        Some(port) => format!("{}://{}:{}", url.scheme(), host, port),
265        None => format!("{}://{}", url.scheme(), host),
266    }
267}
268
269pub(crate) fn referer_for(prev: Option<&str>, current_origin: &str) -> String {
270    let Some(prev) = prev else {
271        return format!("{current_origin}/");
272    };
273    let Ok(parsed) = Url::parse(prev) else {
274        return format!("{current_origin}/");
275    };
276    if parsed.scheme() == "https" && !current_origin.starts_with("https:") {
277        return String::new();
278    }
279    if url_origin(&parsed) != current_origin {
280        let origin = url_origin(&parsed);
281        return format!("{origin}/");
282    }
283    let mut parsed = without_userinfo(parsed);
284    parsed.set_fragment(None);
285    parsed.to_string()
286}
287
288#[cfg(all(test, feature = "http3"))]
289mod alt_svc_tests;
290#[cfg(test)]
291mod redact_tests;
292#[cfg(test)]
293mod referer_tests;
294#[cfg(test)]
295mod reorder_tests;
296
297struct Turn<'a> {
298    code: u16,
299    headers: &'a [(http::HeaderName, http::HeaderValue)],
300    redirect_policy: &'a RedirectPolicy,
301    redirect_cap: usize,
302}
303
304impl Turn<'_> {
305    fn take(
306        &self,
307        journey: &mut Journey,
308        digest: Option<&mut DigestLeg>,
309        replay_body: Option<Body>,
310    ) -> Result<bool> {
311        if let Some(location) = redirect_location(self.code, self.headers)
312            && let Some(hop) = journey.approved_hop(self.redirect_policy, self.code, &location)
313        {
314            journey.follow(hop, self.code, &location, replay_body)?;
315            if journey.chain.len() > self.redirect_cap {
316                return Err(Error::new(Kind::Redirect)
317                    .with_message(format!("too many redirects (max {})", self.redirect_cap)));
318            }
319            if let Some(digest) = digest {
320                journey.authorized = digest.next_hop(journey)?;
321            }
322            return Ok(true);
323        }
324        if self.code == 401
325            && let Some(digest) = digest
326            && let Some(authorized) = digest.answer(journey, self.headers)?
327        {
328            journey.body = replay_body.ok_or_else(unreplayable_body)?;
329            journey.authorized = Some(authorized);
330            return Ok(true);
331        }
332        Ok(false)
333    }
334}