haki_dl/
http.rs

1//! HTTP request and response abstractions.
2
3use std::collections::BTreeMap;
4use std::io::Read;
5use std::sync::{Arc, OnceLock};
6use std::time::Duration;
7
8use brotli::Decompressor;
9use flate2::read::{DeflateDecoder, GzDecoder, ZlibDecoder};
10use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
11use reqwest::redirect::Policy;
12use reqwest::{
13    Client as AsyncClient, RequestBuilder as AsyncRequestBuilder, Response as AsyncResponse,
14};
15
16use crate::cancellation::CancellationToken;
17use crate::config::DownloadOptions;
18use crate::error::{Error, Result};
19
20pub(crate) const HTTP_CONNECTION_POOL_LIMIT: usize = 1024;
21pub(crate) const SOURCE_RETRY_ATTEMPTS: usize = 10;
22pub(crate) const SOURCE_RETRY_DELAY: Duration = Duration::from_millis(1500);
23pub(crate) const SOURCE_RETRY_DELAY_INCREMENT: Duration = Duration::from_millis(0);
24pub(crate) const LIVE_REFRESH_RETRY_ATTEMPTS: usize = 5;
25pub(crate) const LIVE_REFRESH_RETRY_DELAY: Duration = Duration::from_millis(1000);
26
27/// HTTP request model used by transport implementations.
28#[derive(Clone, Debug, Eq, PartialEq)]
29pub struct HttpRequest {
30    /// Request URL.
31    pub url: String,
32    /// Request headers.
33    pub headers: BTreeMap<String, String>,
34}
35
36impl HttpRequest {
37    /// Creates a request model.
38    pub fn new(url: impl Into<String>) -> Self {
39        Self {
40            url: url.into(),
41            headers: BTreeMap::new(),
42        }
43    }
44}
45
46/// HTTP response model used by manifest and segment loaders.
47#[derive(Clone, Debug, Eq, PartialEq)]
48pub struct HttpResponse {
49    /// Numeric status code.
50    pub status: u16,
51    /// Final URL after manual redirect handling.
52    pub final_url: String,
53    /// Response headers.
54    pub headers: BTreeMap<String, String>,
55    /// Response body bytes.
56    pub body: Vec<u8>,
57    /// Debug lines collected while following redirects for source requests.
58    pub debug_logs: Vec<String>,
59}
60
61/// Default async HTTP transport used by manifest and segment loaders.
62#[derive(Clone, Debug)]
63pub struct DefaultHttpClient {
64    timeout: Duration,
65    proxy_mode: ProxyMode,
66    allow_insecure_tls: bool,
67    async_client: Arc<OnceLock<AsyncClient>>,
68}
69
70/// Proxy mode for HTTP transport.
71#[derive(Clone, Debug, Default, Eq, PartialEq)]
72pub enum ProxyMode {
73    /// Use transport/system defaults.
74    #[default]
75    System,
76    /// Do not configure a custom proxy.
77    Disabled,
78    /// Use a custom proxy URL.
79    Custom(String),
80}
81
82impl Default for DefaultHttpClient {
83    fn default() -> Self {
84        Self {
85            timeout: Duration::from_secs(100),
86            proxy_mode: ProxyMode::System,
87            allow_insecure_tls: false,
88            async_client: Arc::new(OnceLock::new()),
89        }
90    }
91}
92
93impl DefaultHttpClient {
94    /// Creates a client with compatibility defaults.
95    pub fn new() -> Self {
96        Self::default()
97    }
98
99    /// Creates a client from public download options.
100    pub fn from_options(options: &DownloadOptions) -> Self {
101        let proxy_mode = match (&options.custom_proxy, options.use_system_proxy) {
102            (Some(proxy), _) => ProxyMode::Custom(proxy.clone()),
103            (None, true) => ProxyMode::System,
104            (None, false) => ProxyMode::Disabled,
105        };
106        Self::new()
107            .with_timeout(options.http_request_timeout)
108            .with_proxy_mode(proxy_mode)
109            .with_insecure_tls(options.allow_insecure_tls)
110    }
111
112    /// Overrides the request timeout.
113    pub fn with_timeout(mut self, timeout: Duration) -> Self {
114        self.timeout = timeout;
115        self.reset_clients();
116        self
117    }
118
119    /// Overrides proxy behavior.
120    pub fn with_proxy_mode(mut self, proxy_mode: ProxyMode) -> Self {
121        self.proxy_mode = proxy_mode;
122        self.reset_clients();
123        self
124    }
125
126    /// Enables or disables invalid TLS certificate acceptance.
127    pub fn with_insecure_tls(mut self, allow: bool) -> Self {
128        self.allow_insecure_tls = allow;
129        self.reset_clients();
130        self
131    }
132
133    /// Returns configured timeout.
134    pub fn timeout(&self) -> Duration {
135        self.timeout
136    }
137
138    /// Returns configured proxy mode.
139    pub fn proxy_mode(&self) -> &ProxyMode {
140        &self.proxy_mode
141    }
142
143    /// Returns whether invalid TLS certificates are accepted.
144    pub fn allow_insecure_tls(&self) -> bool {
145        self.allow_insecure_tls
146    }
147
148    /// Sends one GET request with manual redirect handling.
149    pub async fn send(&self, request: HttpRequest) -> Result<HttpResponse> {
150        let client = self.async_client()?;
151        let mut current_url = request.url.clone();
152        let mut debug_logs = Vec::new();
153        loop {
154            let response = self
155                .single_get(&client, &current_url, &request.headers)
156                .await?;
157            let status = response.status().as_u16();
158            if is_redirect(status)
159                && let Some(location) = response
160                    .headers()
161                    .get(reqwest::header::LOCATION)
162                    .and_then(|value| value.to_str().ok())
163            {
164                let redirected = resolve_redirect(&current_url, location);
165                if redirected != current_url {
166                    debug_logs.push(format_response_headers(response.headers()));
167                    debug_logs.push(format!("Fetch: {redirected}"));
168                    debug_logs.push(format_source_http_request_headers(&request.headers));
169                    current_url = redirected;
170                    let _ = response.bytes().await;
171                    continue;
172                }
173            }
174            if !(200..=299).contains(&status) {
175                let _ = response.bytes().await;
176                return Err(Error::http(format!(
177                    "HTTP status {status} for {current_url}"
178                )));
179            }
180            let headers = response_headers(response.headers());
181            let body = response
182                .bytes()
183                .await
184                .map_err(|error| Error::http(self.redact_error(error.to_string())))?
185                .to_vec();
186            let body = decode_content_body(body, &headers)?;
187            return Ok(HttpResponse {
188                status,
189                final_url: current_url,
190                headers,
191                body,
192                debug_logs,
193            });
194        }
195    }
196
197    /// Sends one source-probe GET without draining MPEG-TS live streams.
198    pub async fn send_source(&self, request: HttpRequest) -> Result<HttpResponse> {
199        let client = self.async_client()?;
200        let mut current_url = request.url.clone();
201        let mut debug_logs = Vec::new();
202        loop {
203            let response = self
204                .single_get(&client, &current_url, &request.headers)
205                .await?;
206            let status = response.status().as_u16();
207            if is_redirect(status)
208                && let Some(location) = response
209                    .headers()
210                    .get(reqwest::header::LOCATION)
211                    .and_then(|value| value.to_str().ok())
212            {
213                let redirected = resolve_redirect(&current_url, location);
214                if redirected != current_url {
215                    debug_logs.push(format_response_headers(response.headers()));
216                    debug_logs.push(format!("Fetch: {redirected}"));
217                    debug_logs.push(format_source_http_request_headers(&request.headers));
218                    current_url = redirected;
219                    let _ = response.bytes().await;
220                    continue;
221                }
222            }
223            if !(200..=299).contains(&status) {
224                let _ = response.bytes().await;
225                return Err(Error::http(format!(
226                    "HTTP status {status} for {current_url}"
227                )));
228            }
229            let headers = response_headers(response.headers());
230            let body = read_source_probe_body(response, self).await?;
231            let body = decode_content_body(body, &headers)?;
232            return Ok(HttpResponse {
233                status,
234                final_url: current_url,
235                headers,
236                body,
237                debug_logs,
238            });
239        }
240    }
241
242    fn reset_clients(&mut self) {
243        self.async_client = Arc::new(OnceLock::new());
244    }
245
246    fn async_client(&self) -> Result<AsyncClient> {
247        if let Some(client) = self.async_client.get() {
248            return Ok(client.clone());
249        }
250        let client = match &self.proxy_mode {
251            ProxyMode::System => {
252                shared_http_client(self.timeout, None, true, self.allow_insecure_tls)
253            }
254            ProxyMode::Disabled => {
255                shared_http_client(self.timeout, None, false, self.allow_insecure_tls)
256            }
257            ProxyMode::Custom(proxy) => {
258                shared_http_client(self.timeout, Some(proxy), false, self.allow_insecure_tls)
259            }
260        }?;
261        let _ = self.async_client.set(client.clone());
262        Ok(client)
263    }
264
265    async fn single_get(
266        &self,
267        client: &AsyncClient,
268        url: &str,
269        headers: &BTreeMap<String, String>,
270    ) -> Result<AsyncResponse> {
271        let mut request = client
272            .get(url)
273            .header("Accept-Encoding", "gzip, deflate")
274            .header("Cache-Control", "no-cache");
275        request = apply_request_headers(request, headers);
276        request
277            .send()
278            .await
279            .map_err(|error| Error::http(self.redact_error(error.to_string())))
280    }
281
282    fn redact_error(&self, message: String) -> String {
283        match &self.proxy_mode {
284            ProxyMode::Custom(proxy) => redact_proxy_error_message(proxy, &message),
285            ProxyMode::System | ProxyMode::Disabled => message,
286        }
287    }
288}
289
290fn build_http_client_inner(
291    timeout: Duration,
292    custom_proxy: Option<&str>,
293    use_system_proxy: bool,
294    allow_insecure_tls: bool,
295) -> Result<AsyncClient> {
296    let mut builder = AsyncClient::builder()
297        .redirect(Policy::none())
298        .timeout(timeout)
299        .connect_timeout(timeout)
300        .pool_max_idle_per_host(HTTP_CONNECTION_POOL_LIMIT)
301        .danger_accept_invalid_certs(allow_insecure_tls);
302    if !use_system_proxy {
303        builder = builder.no_proxy();
304    }
305    if let Some(proxy) = custom_proxy {
306        let proxy = reqwest::Proxy::all(proxy).map_err(|error| {
307            Error::config(redact_proxy_error_message(proxy, &error.to_string()))
308        })?;
309        builder = builder.proxy(proxy);
310    }
311    builder.build().map_err(|error| {
312        let message = match custom_proxy {
313            Some(proxy) => redact_proxy_error_message(proxy, &error.to_string()),
314            None => error.to_string(),
315        };
316        Error::http(message)
317    })
318}
319
320pub(crate) fn shared_http_client(
321    timeout: Duration,
322    custom_proxy: Option<&str>,
323    use_system_proxy: bool,
324    allow_insecure_tls: bool,
325) -> Result<AsyncClient> {
326    build_http_client_inner(timeout, custom_proxy, use_system_proxy, allow_insecure_tls)
327}
328
329impl crate::traits::HttpClient for DefaultHttpClient {
330    fn send(
331        &self,
332        request: HttpRequest,
333    ) -> impl std::future::Future<Output = Result<HttpResponse>> + Send + '_ {
334        self.send(request)
335    }
336}
337
338pub(crate) async fn sleep_for_retry(
339    duration: Duration,
340    cancellation_token: Option<&CancellationToken>,
341) -> Result<()> {
342    let Some(cancellation_token) = cancellation_token else {
343        tokio::time::sleep(duration).await;
344        return Ok(());
345    };
346    let started = std::time::Instant::now();
347    loop {
348        cancellation_token.check()?;
349        let elapsed = started.elapsed();
350        if elapsed >= duration {
351            return Ok(());
352        }
353        tokio::time::sleep(
354            duration
355                .saturating_sub(elapsed)
356                .min(Duration::from_millis(50)),
357        )
358        .await;
359    }
360}
361
362pub(crate) fn apply_request_headers(
363    mut request: AsyncRequestBuilder,
364    headers: &BTreeMap<String, String>,
365) -> AsyncRequestBuilder {
366    for (key, value) in headers {
367        let Ok(name) = HeaderName::from_bytes(key.as_bytes()) else {
368            continue;
369        };
370        let Ok(value) = HeaderValue::from_str(value) else {
371            continue;
372        };
373        request = request.header(name, value);
374    }
375    request
376}
377
378fn redact_proxy_error_message(proxy: &str, message: &str) -> String {
379    message.replace(proxy, &redact_proxy_url(proxy))
380}
381
382fn redact_proxy_url(value: &str) -> String {
383    let Some(scheme_end) = value.find("://") else {
384        return value.to_string();
385    };
386    let authority_start = scheme_end + 3;
387    let Some(at_offset) = value[authority_start..].find('@') else {
388        return value.to_string();
389    };
390    let at = authority_start + at_offset;
391    format!("{}***@{}", &value[..authority_start], &value[at + 1..])
392}
393
394fn is_redirect(status: u16) -> bool {
395    (300..=399).contains(&status)
396}
397
398fn response_headers(headers_map: &HeaderMap) -> BTreeMap<String, String> {
399    let mut headers = BTreeMap::new();
400    for (name, value) in headers_map {
401        if let Ok(value) = value.to_str() {
402            headers.insert(name.as_str().to_ascii_lowercase(), value.to_string());
403        }
404    }
405    headers
406}
407
408fn format_response_headers(headers: &HeaderMap) -> String {
409    let mut text = String::new();
410    for (name, value) in headers {
411        if let Ok(value) = value.to_str() {
412            text.push_str(name.as_str());
413            text.push_str(": ");
414            text.push_str(value);
415            text.push('\n');
416        }
417    }
418    text.trim_end().to_string()
419}
420
421fn format_source_http_request_headers(headers: &BTreeMap<String, String>) -> String {
422    let mut lines = vec![
423        "Accept-Encoding: gzip, deflate".to_string(),
424        "Cache-Control: no-cache".to_string(),
425    ];
426    lines.extend(headers.iter().map(|(key, value)| format!("{key}: {value}")));
427    lines.join("\n")
428}
429
430fn decode_content_body(body: Vec<u8>, headers: &BTreeMap<String, String>) -> Result<Vec<u8>> {
431    let Some(encoding) = headers.get("content-encoding") else {
432        return Ok(body);
433    };
434    let mut decoded = body;
435    for value in encoding
436        .split(',')
437        .map(|value| value.trim().to_ascii_lowercase())
438    {
439        decoded = match value.as_str() {
440            "" | "identity" => decoded,
441            "gzip" if decoded.starts_with(&[0x1f, 0x8b]) => {
442                decode_with(GzDecoder::new(&decoded[..]))?
443            }
444            "gzip" => decoded,
445            "deflate" => decode_deflate(decoded)?,
446            "br" => match decode_with(Decompressor::new(&decoded[..], 4096)) {
447                Ok(decoded_body) => decoded_body,
448                Err(_) => decoded,
449            },
450            _ => decoded,
451        };
452    }
453    Ok(decoded)
454}
455
456fn decode_deflate(body: Vec<u8>) -> Result<Vec<u8>> {
457    match decode_with(ZlibDecoder::new(&body[..])) {
458        Ok(decoded) => Ok(decoded),
459        Err(_) => decode_with(DeflateDecoder::new(&body[..])).or(Ok(body)),
460    }
461}
462
463fn decode_with(mut reader: impl Read) -> Result<Vec<u8>> {
464    let mut decoded = Vec::new();
465    reader.read_to_end(&mut decoded)?;
466    Ok(decoded)
467}
468
469async fn read_source_probe_body(
470    mut response: AsyncResponse,
471    transport: &DefaultHttpClient,
472) -> Result<Vec<u8>> {
473    const PROBE_SIZE: usize = 188 * 5;
474    let mut body = Vec::new();
475    while body.len() < PROBE_SIZE {
476        let Some(chunk) = response
477            .chunk()
478            .await
479            .map_err(|error| Error::http(transport.redact_error(error.to_string())))?
480        else {
481            return Ok(body);
482        };
483        let remaining = PROBE_SIZE.saturating_sub(body.len());
484        let take = remaining.min(chunk.len());
485        body.extend_from_slice(&chunk[..take]);
486        if source_probe_is_mpeg2_ts(&body) || source_probe_looks_binary(&body) {
487            return Ok(body);
488        }
489        if take < chunk.len() {
490            body.extend_from_slice(&chunk[take..]);
491            break;
492        }
493    }
494    while let Some(chunk) = response
495        .chunk()
496        .await
497        .map_err(|error| Error::http(transport.redact_error(error.to_string())))?
498    {
499        body.extend_from_slice(&chunk);
500    }
501    Ok(body)
502}
503
504fn source_probe_is_mpeg2_ts(buffer: &[u8]) -> bool {
505    const PACKET_SIZE: usize = 188;
506    if buffer.len() < PACKET_SIZE {
507        return false;
508    }
509    let packet_count = std::cmp::min(buffer.len() / PACKET_SIZE, 5);
510    let sync_count = (0..packet_count)
511        .filter(|index| buffer[index * PACKET_SIZE] == 0x47)
512        .count();
513    sync_count >= 3
514}
515
516fn source_probe_looks_binary(data: &[u8]) -> bool {
517    if data.is_empty() {
518        return false;
519    }
520    let mut non_text = 0_usize;
521    let mut total = 0_usize;
522    let mut index = 0_usize;
523    while index < data.len() {
524        let byte = data[index];
525        total += 1;
526        if byte == 0 {
527            return true;
528        }
529        if (0x20..=0x7e).contains(&byte) || matches!(byte, 0x09 | 0x0a | 0x0d) {
530            index += 1;
531            continue;
532        }
533        let seq_len = source_probe_utf8_sequence_length(byte);
534        if seq_len > 1
535            && index + seq_len <= data.len()
536            && source_probe_valid_utf8_sequence(&data[index..index + seq_len])
537        {
538            index += seq_len;
539            continue;
540        }
541        non_text += 1;
542        index += 1;
543    }
544    (non_text as f64 / total as f64) > 0.3
545}
546
547fn source_probe_utf8_sequence_length(byte: u8) -> usize {
548    if byte & 0x80 == 0x00 {
549        1
550    } else if byte & 0xe0 == 0xc0 {
551        2
552    } else if byte & 0xf0 == 0xe0 {
553        3
554    } else if byte & 0xf8 == 0xf0 {
555        4
556    } else {
557        1
558    }
559}
560
561fn source_probe_valid_utf8_sequence(seq: &[u8]) -> bool {
562    if seq.len() <= 1 {
563        return false;
564    }
565    seq.iter().skip(1).all(|byte| byte & 0xc0 == 0x80)
566}
567
568fn resolve_redirect(base: &str, location: &str) -> String {
569    if let Ok(base_url) = reqwest::Url::parse(base)
570        && let Ok(joined) = base_url.join(location)
571    {
572        return joined.to_string();
573    }
574    if reqwest::Url::parse(location).is_ok() {
575        return location.to_string();
576    }
577    let Some((origin, base_path)) = split_url_origin_and_path(base) else {
578        return location.to_string();
579    };
580    let (location_path, suffix) = split_path_suffix(location);
581    let joined = if location_path.starts_with('/') {
582        normalize_url_path(location_path)
583    } else {
584        let base_dir = base_path
585            .rsplit_once('/')
586            .map(|(prefix, _)| format!("{prefix}/"))
587            .unwrap_or_else(|| "/".to_string());
588        normalize_url_path(&format!("{base_dir}{location_path}"))
589    };
590    format!("{origin}{joined}{suffix}")
591}
592
593fn split_url_origin_and_path(url: &str) -> Option<(&str, &str)> {
594    let scheme_end = url.find("://")?;
595    let after_scheme = scheme_end + 3;
596    let rest = url.get(after_scheme..)?;
597    let path_start = rest.find('/').map(|index| after_scheme + index);
598    let index = path_start.unwrap_or(url.len());
599    let origin = url.get(..index)?;
600    let path = url.get(index..).unwrap_or("/");
601    Some((origin, path))
602}
603
604fn split_path_suffix(value: &str) -> (&str, &str) {
605    let query = value.find('?');
606    let fragment = value.find('#');
607    let split = match (query, fragment) {
608        (Some(left), Some(right)) => left.min(right),
609        (Some(index), None) | (None, Some(index)) => index,
610        (None, None) => value.len(),
611    };
612    (&value[..split], &value[split..])
613}
614
615fn normalize_url_path(path: &str) -> String {
616    let mut parts = Vec::new();
617    for part in path.split('/') {
618        match part {
619            "" | "." => {}
620            ".." => {
621                let _ = parts.pop();
622            }
623            value => parts.push(value),
624        }
625    }
626    format!("/{}", parts.join("/"))
627}
haki_dl/http.rs

haki_dl/
http.rs
