subscription_proxy_pool/
subscription.rs

1//! Fetch and parse subscriptions without exposing their URLs in diagnostics.
2
3use std::{collections::HashSet, fmt, net::Ipv6Addr};
4
5use base64::{Engine, engine::general_purpose};
6use reqwest::header::{ETAG, HeaderValue, IF_MODIFIED_SINCE, IF_NONE_MATCH, LAST_MODIFIED};
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use sha2::{Digest, Sha256};
10use url::Url;
11
12use crate::{Error, ProxyNode, Result};
13
14/// Default limit for both downloaded and decoded subscription documents (4 MiB).
15pub const DEFAULT_MAX_SUBSCRIPTION_BYTES: usize = 4 * 1024 * 1024;
16
17/// Resource limits for parsing an untrusted subscription document.
18#[derive(Clone, Debug)]
19pub struct ParseOptions {
20    /// Maximum size of the input and decoded document, in bytes. Must be positive.
21    pub max_bytes: usize,
22}
23
24impl Default for ParseOptions {
25    fn default() -> Self {
26        Self {
27            max_bytes: DEFAULT_MAX_SUBSCRIPTION_BYTES,
28        }
29    }
30}
31
32/// HTTP validators belonging to a successfully parsed subscription response.
33///
34/// Keep these with the corresponding nodes and only reuse them for the same
35/// [`SubscriptionSource`]. Empty values and invalid HTTP header values are rejected
36/// before sending a request. Servers determine whether validators still match.
37#[derive(Clone, Debug, Default, Serialize, Deserialize)]
38pub struct SubscriptionValidators {
39    /// The response's `ETag`, including quotes and an optional `W/` prefix.
40    pub etag: Option<String>,
41    /// The response's `Last-Modified` HTTP date.
42    pub last_modified: Option<String>,
43}
44
45impl SubscriptionValidators {
46    pub(crate) fn is_valid(&self) -> bool {
47        [self.etag.as_deref(), self.last_modified.as_deref()]
48            .into_iter()
49            .flatten()
50            .all(|value| validator_header(value).is_some())
51    }
52}
53
54/// A conditional subscription request's result.
55#[derive(Clone, Debug)]
56pub enum SubscriptionUpdate {
57    /// New content was received and successfully parsed.
58    Modified {
59        /// Nodes and skipped-entry count from the new content.
60        report: ParseReport,
61        /// Validators from this response, replacing any previous validators.
62        validators: SubscriptionValidators,
63    },
64    /// HTTP 304: retain the nodes belonging to the supplied validators.
65    NotModified,
66}
67
68/// Usable nodes, in subscription order, and the number of discarded entries.
69#[derive(Clone, Debug)]
70pub struct ParseReport {
71    /// Valid nodes with duplicate endpoints removed. The first name wins.
72    pub nodes: Vec<ProxyNode>,
73    /// Unsupported, malformed, or duplicate entries. Blank lines and comments do not count.
74    pub skipped: usize,
75}
76
77/// A subscription URL whose debug representation reveals only a stable hash.
78#[derive(Clone)]
79pub struct SubscriptionSource {
80    url: Url,
81}
82
83impl SubscriptionSource {
84    /// Accept an HTTP(S) URL. Query parameters and credentials remain private.
85    pub fn new(value: &str) -> Result<Self> {
86        let mut url = Url::parse(value).map_err(|_| Error::Config("invalid subscription URL"))?;
87        if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none_or(str::is_empty) {
88            return Err(Error::Config("subscription URL must use HTTP or HTTPS"));
89        }
90        // Fragments are never transmitted and must not create separate cache entries.
91        url.set_fragment(None);
92        Ok(Self { url })
93    }
94
95    /// SHA-256 of the normalized URL, suitable for isolating subscription caches.
96    pub fn key(&self) -> String {
97        format!("{:x}", Sha256::digest(self.url.as_str().as_bytes()))
98    }
99
100    /// Download and parse a subscription with a streaming response-size limit.
101    ///
102    /// `max_bytes` must be positive. The supplied client's timeout,
103    /// redirect, and proxy policies apply. HTTP failure responses are rejected.
104    pub async fn fetch(&self, client: &reqwest::Client, max_bytes: usize) -> Result<ParseReport> {
105        match self.fetch_update(client, max_bytes, None).await? {
106            SubscriptionUpdate::Modified { report, .. } => Ok(report),
107            // fetch_update rejects unsolicited 304 responses. Keep the same
108            // checked error here if its implementation changes in the future.
109            SubscriptionUpdate::NotModified => Err(Error::Subscription(
110                "received HTTP 304 without subscription validators",
111            )),
112        }
113    }
114
115    /// Download a subscription, optionally revalidating previously obtained nodes.
116    ///
117    /// Sends `If-None-Match` and/or `If-Modified-Since` from `validators`. A 304
118    /// response succeeds only when at least one validator was actually sent; it
119    /// has no body to parse. Callers must retain the matching nodes themselves.
120    /// A modified response replaces the old validators, including clearing any
121    /// that the server no longer sends. No automatic retries are performed.
122    ///
123    /// `max_bytes` is a positive limit on the downloaded and decoded document.
124    /// The supplied client's timeout, redirect, and proxy policies apply.
125    pub async fn fetch_update(
126        &self,
127        client: &reqwest::Client,
128        max_bytes: usize,
129        validators: Option<&SubscriptionValidators>,
130    ) -> Result<SubscriptionUpdate> {
131        if max_bytes == 0 {
132            return Err(Error::Config("subscription size limit must be positive"));
133        }
134        let mut request = client.get(self.url.clone());
135        let mut conditional = false;
136        if let Some(validators) = validators {
137            for (name, value) in [
138                (IF_NONE_MATCH, validators.etag.as_deref()),
139                (IF_MODIFIED_SINCE, validators.last_modified.as_deref()),
140            ] {
141                if let Some(value) = value {
142                    let value = validator_header(value)
143                        .ok_or(Error::Config("invalid subscription validator"))?;
144                    request = request.header(name, value);
145                    conditional = true;
146                }
147            }
148        }
149        let response = request
150            .send()
151            .await
152            .map_err(|error| Error::Transport(error.without_url()))?;
153        if response.status() == reqwest::StatusCode::NOT_MODIFIED {
154            return if conditional {
155                Ok(SubscriptionUpdate::NotModified)
156            } else {
157                Err(Error::Subscription(
158                    "received HTTP 304 without subscription validators",
159                ))
160            };
161        }
162        let mut response = response
163            .error_for_status()
164            .map_err(|error| Error::Transport(error.without_url()))?;
165        if response
166            .content_length()
167            .is_some_and(|size| size > max_bytes as u64)
168        {
169            return Err(Error::Subscription("subscription exceeds size limit"));
170        }
171        let validators = SubscriptionValidators {
172            etag: response.headers().get(ETAG).and_then(response_validator),
173            last_modified: response
174                .headers()
175                .get(LAST_MODIFIED)
176                .and_then(response_validator),
177        };
178        let mut body = Vec::new();
179        while let Some(chunk) = response
180            .chunk()
181            .await
182            .map_err(|error| Error::Transport(error.without_url()))?
183        {
184            if chunk.len() > max_bytes.saturating_sub(body.len()) {
185                return Err(Error::Subscription("subscription exceeds size limit"));
186            }
187            body.extend_from_slice(&chunk);
188        }
189        let content = std::str::from_utf8(&body)
190            .map_err(|_| Error::Subscription("subscription is not UTF-8"))?;
191        let report = parse_subscription_with_options(content, &ParseOptions { max_bytes })?;
192        Ok(SubscriptionUpdate::Modified { report, validators })
193    }
194}
195
196fn validator_header(value: &str) -> Option<HeaderValue> {
197    if value.trim().is_empty() {
198        return None;
199    }
200    HeaderValue::from_str(value).ok()
201}
202
203fn response_validator(value: &HeaderValue) -> Option<String> {
204    let value = value.to_str().ok()?;
205    validator_header(value).map(|_| value.to_owned())
206}
207
208impl fmt::Debug for SubscriptionSource {
209    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
210        formatter
211            .debug_struct("SubscriptionSource")
212            .field("key", &self.key())
213            .finish_non_exhaustive()
214    }
215}
216
217#[derive(Deserialize)]
218struct ClashSubscription {
219    proxies: Vec<Value>,
220}
221
222/// Parse Clash YAML or newline-separated HTTP(S)/SOCKS5(H) proxy URLs.
223///
224/// Either format may be wrapped once in standard or URL-safe Base64, with or
225/// without padding or whitespace. Unsupported protocols, malformed entries,
226/// and duplicates are skipped; an empty usable result is an error. No parsing
227/// error includes subscription content or credentials.
228pub fn parse_subscription(content: &str) -> Result<ParseReport> {
229    parse_subscription_with_options(content, &ParseOptions::default())
230}
231
232/// Parse a subscription using an explicit input and decoded-document size limit.
233///
234/// This has the same format and error behavior as [`parse_subscription`], with
235/// a caller-selected limit instead of the default 4 MiB. YAML also retains the
236/// parser's structural limits on nesting and alias expansion.
237pub fn parse_subscription_with_options(
238    content: &str,
239    options: &ParseOptions,
240) -> Result<ParseReport> {
241    if options.max_bytes == 0 {
242        return Err(Error::Config("subscription size limit must be positive"));
243    }
244    if content.len() > options.max_bytes {
245        return Err(Error::Subscription("subscription exceeds size limit"));
246    }
247    let content = content.trim().trim_start_matches('\u{feff}').trim();
248    if content.is_empty() {
249        return Err(Error::Subscription("subscription is empty"));
250    }
251    let mut parse_error = match parse_plain(content) {
252        Ok(report) => return Ok(report),
253        Err(error) => error,
254    };
255    let encoded: String = content
256        .chars()
257        .filter(|value| !value.is_whitespace())
258        .collect();
259    for engine in [
260        &general_purpose::STANDARD,
261        &general_purpose::STANDARD_NO_PAD,
262        &general_purpose::URL_SAFE,
263        &general_purpose::URL_SAFE_NO_PAD,
264    ] {
265        let Ok(decoded) = engine.decode(&encoded) else {
266            continue;
267        };
268        // Base64 cannot expand beyond its input size, but keep the decoded
269        // document's invariant explicit if decoding changes in the future.
270        if decoded.len() > options.max_bytes {
271            return Err(Error::Subscription("subscription exceeds size limit"));
272        }
273        let Ok(decoded) = std::str::from_utf8(&decoded) else {
274            continue;
275        };
276        match parse_plain(decoded.trim().trim_start_matches('\u{feff}').trim()) {
277            Ok(report) => return Ok(report),
278            Err(error) => parse_error = error,
279        }
280    }
281    Err(parse_error)
282}
283
284fn parse_plain(content: &str) -> Result<ParseReport> {
285    // An initial URI identifies a line list, even if a later malformed entry
286    // resembles a YAML key. Decide this before looking for Clash diagnostics,
287    // so one bad line cannot discard the list's usable nodes.
288    let starts_with_uri = content
289        .lines()
290        .map(str::trim)
291        .find(|line| !line.is_empty() && !line.starts_with('#'))
292        .is_some_and(|line| Url::parse(line).is_ok_and(|url| url.has_host()));
293    if starts_with_uri {
294        return collect_uri_nodes(content);
295    }
296    if let Ok(clash) = serde_saphyr::from_str::<ClashSubscription>(content) {
297        return collect_nodes(clash.proxies.iter().map(clash_node));
298    }
299    // A recognizable Clash document with an invalid structure is a parsing
300    // failure, rather than a list whose nodes happen to use unsupported protocols.
301    // Retain only a static diagnostic: YAML parser errors can contain credentials.
302    if content.lines().any(|line| {
303        line.trim_start()
304            .strip_prefix("proxies")
305            .and_then(|rest| rest.trim_start().strip_prefix(':'))
306            .is_some_and(|rest| {
307                rest.is_empty()
308                    || rest.starts_with(char::is_whitespace)
309                    || rest.starts_with(['[', '{'])
310            })
311    }) {
312        return Err(Error::Subscription("invalid Clash subscription document"));
313    }
314    collect_uri_nodes(content)
315}
316
317fn collect_uri_nodes(content: &str) -> Result<ParseReport> {
318    collect_nodes(
319        content
320            .lines()
321            .map(str::trim)
322            .filter(|line| !line.is_empty() && !line.starts_with('#'))
323            .map(|line| ProxyNode::from_url(line).ok()),
324    )
325}
326
327fn collect_nodes(nodes: impl Iterator<Item = Option<ProxyNode>>) -> Result<ParseReport> {
328    let mut report = ParseReport {
329        nodes: Vec::new(),
330        skipped: 0,
331    };
332    let mut seen = HashSet::new();
333    for node in nodes {
334        match node {
335            Some(node) if seen.insert(node.id()) => report.nodes.push(node),
336            _ => report.skipped += 1,
337        }
338    }
339    if report.nodes.is_empty() {
340        return Err(Error::Subscription(
341            "subscription contains no supported valid proxy nodes",
342        ));
343    }
344    Ok(report)
345}
346
347fn clash_node(value: &Value) -> Option<ProxyNode> {
348    let value = value.as_object()?;
349    let protocol = value.get("type")?.as_str()?;
350    let tls = match value.get("tls") {
351        Some(value) => value.as_bool()?,
352        None => false,
353    };
354    let scheme = match (protocol, tls) {
355        ("http", false) => "http",
356        ("http", true) | ("https", _) => "https",
357        ("socks5", false) => "socks5",
358        ("socks5h", false) => "socks5h",
359        // TLS-wrapped SOCKS is not plain SOCKS and cannot be represented safely.
360        _ => return None,
361    };
362    let server = value.get("server")?.as_str()?;
363    if server.is_empty() {
364        return None;
365    }
366    let port = match value.get("port")? {
367        Value::Number(number) => u16::try_from(number.as_u64()?).ok()?,
368        Value::String(port) => port.parse::<u16>().ok()?,
369        _ => return None,
370    };
371    if port == 0 {
372        return None;
373    }
374    let mut url = Url::parse(&format!("{scheme}://localhost")).ok()?;
375    let host = if server.parse::<Ipv6Addr>().is_ok() {
376        format!("[{server}]")
377    } else {
378        server.to_owned()
379    };
380    // Unlike Host::parse, set_host accepts a trailing ':port' and truncates it.
381    // Validate the whole server field before installing it into the URL.
382    let host = url::Host::parse(&host).ok()?;
383    url.set_host(Some(&host.to_string())).ok()?;
384    url.set_port(Some(port)).ok()?;
385    if let Some(username) = value.get("username") {
386        // URL setters preserve existing percent escapes. Clash credentials are
387        // literal strings, so escape '%' first to prevent accidental decoding.
388        url.set_username(&username.as_str()?.replace('%', "%25"))
389            .ok()?;
390    }
391    if let Some(password) = value.get("password") {
392        url.set_password(Some(&password.as_str()?.replace('%', "%25")))
393            .ok()?;
394    }
395    let mut node = ProxyNode::from_url(url.as_str()).ok()?;
396    if let Some(name) = value.get("name") {
397        node = node.with_name(name.as_str()?);
398    }
399    Some(node)
400}
401
402#[cfg(test)]
403mod tests {
404    use std::time::Duration;
405
406    use tokio::io::{AsyncReadExt, AsyncWriteExt};
407
408    use super::*;
409    use crate::ProxyKind;
410
411    #[test]
412    fn clash_preserves_protocol_auth_and_ipv6_and_skips_bad_entries() {
413        let report = parse_subscription(r#"
414proxies:
415  - {name: secure, type: http, server: proxy.test, port: 8443, tls: true, username: 'a@b', password: 'p:/?#@'}
416  - {name: ipv6, type: socks5h, server: '::1', port: '1080'}
417  - {type: ss, server: proxy.test, port: 8388, password: hidden}
418  - {type: socks5, server: proxy.test, port: 1080, tls: true}
419  - {type: http, server: proxy.test, port: -1}
420  - {type: http, server: proxy.test}
421  - invalid
422"#).unwrap();
423        assert_eq!(report.nodes.len(), 2);
424        assert_eq!(report.skipped, 5);
425        assert_eq!(report.nodes[0].kind(), ProxyKind::Https);
426        assert_eq!(report.nodes[0].name(), "secure");
427        let url = Url::parse(report.nodes[0].url()).unwrap();
428        assert_eq!(url.username(), "a%40b");
429        assert_eq!(url.password(), Some("p%3A%2F%3F%23%40"));
430        assert_eq!(report.nodes[1].url(), "socks5h://[::1]:1080");
431    }
432
433    #[test]
434    fn uri_lists_deduplicate_in_order_and_ignore_comments() {
435        let report = parse_subscription("\u{feff}# comment\nhttp://host:80#first\nss://unsupported\n\nhttps://host:443\nhttp://host#second\n").unwrap();
436        assert_eq!(report.nodes.len(), 2);
437        assert_eq!(report.nodes[0].name(), "first");
438        assert_eq!(report.nodes[1].kind(), ProxyKind::Https);
439        assert_eq!(report.skipped, 2);
440    }
441
442    #[test]
443    fn literal_percent_credentials_are_preserved_and_invalid_hosts_are_skipped() {
444        let report = parse_subscription(
445            r#"
446proxies:
447  - {type: http, server: host, port: 8080, username: 'user%40literal', password: 'pass%2Fword'}
448  - {type: http, server: 'host:9999', port: 8080}
449  - {type: http, server: 'host/path', port: 8080}
450"#,
451        )
452        .unwrap();
453        assert_eq!(report.nodes.len(), 1);
454        assert_eq!(report.skipped, 2);
455        let url = Url::parse(report.nodes[0].url()).unwrap();
456        assert_eq!(url.username(), "user%2540literal");
457        assert_eq!(url.password(), Some("pass%252Fword"));
458    }
459
460    #[test]
461    fn accepts_base64_engines_padding_and_whitespace() {
462        for content in [
463            "http://user:secret@host:8080#节点??\nsocks5://host:1080",
464            "proxies:\n  - {type: http, server: host, port: 8080}",
465        ] {
466            for engine in [
467                &general_purpose::STANDARD,
468                &general_purpose::STANDARD_NO_PAD,
469                &general_purpose::URL_SAFE,
470                &general_purpose::URL_SAFE_NO_PAD,
471            ] {
472                let encoded = engine.encode(content);
473                let spaced = encoded
474                    .chars()
475                    .enumerate()
476                    .fold(String::new(), |mut out, (i, ch)| {
477                        if i % 13 == 0 {
478                            out.push_str(" \n\t");
479                        }
480                        out.push(ch);
481                        out
482                    });
483                assert_eq!(
484                    parse_subscription(&spaced).unwrap().nodes.len(),
485                    parse_subscription(content).unwrap().nodes.len()
486                );
487            }
488        }
489    }
490
491    #[test]
492    fn parsing_errors_and_debug_do_not_expose_secrets() {
493        for content in [
494            "",
495            "proxies: []",
496            "ss://private:secret@host",
497            "proxies: [secret",
498        ] {
499            let error = parse_subscription(content).unwrap_err();
500            assert!(!format!("{error:?} {error}").contains("secret"));
501        }
502        assert!(parse_subscription(&" ".repeat(DEFAULT_MAX_SUBSCRIPTION_BYTES + 1)).is_err());
503        let source =
504            SubscriptionSource::new("https://private:secret@host/private-path?token=hidden")
505                .unwrap();
506        let debug = format!("{source:?}");
507        for secret in ["private", "secret", "hidden", "host"] {
508            assert!(!debug.contains(secret));
509        }
510        assert_eq!(source.key().len(), 64);
511        assert_ne!(
512            source.key(),
513            SubscriptionSource::new("https://host/other").unwrap().key()
514        );
515        assert!(SubscriptionSource::new("file:///secret").is_err());
516        assert!(SubscriptionSource::new("invalid-secret").is_err());
517    }
518
519    async fn serve(response: &'static str) -> (SubscriptionSource, tokio::task::JoinHandle<()>) {
520        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
521        let address = listener.local_addr().unwrap();
522        let server = tokio::spawn(async move {
523            let (mut socket, _) = listener.accept().await.unwrap();
524            let mut request = [0; 4096];
525            let _ = socket.read(&mut request).await.unwrap();
526            socket.write_all(response.as_bytes()).await.unwrap();
527        });
528        (
529            SubscriptionSource::new(&format!("http://{address}/private?token=secret")).unwrap(),
530            server,
531        )
532    }
533
534    fn client() -> reqwest::Client {
535        reqwest::Client::builder()
536            .no_proxy()
537            .timeout(Duration::from_secs(3))
538            .build()
539            .unwrap()
540    }
541
542    #[tokio::test]
543    async fn fetch_checks_status_and_content_length_without_leaking_url() {
544        let (source, server) =
545            serve("HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await;
546        let error = source.fetch(&client(), 1024).await.unwrap_err();
547        assert!(!format!("{error} {error:?}").contains("secret"));
548        assert!(!format!("{error} {error:?}").contains("private"));
549        server.await.unwrap();
550
551        let (source, server) =
552            serve("HTTP/1.1 200 OK\r\nContent-Length: 1000\r\nConnection: close\r\n\r\n").await;
553        assert!(matches!(
554            source.fetch(&client(), 20).await,
555            Err(Error::Subscription("subscription exceeds size limit"))
556        ));
557        server.await.unwrap();
558    }
559
560    #[tokio::test]
561    async fn fetch_checks_chunked_response_size_and_parses_success() {
562        let (source, server) = serve("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n11\r\nhttp://host:8080\n\r\n0\r\n\r\n").await;
563        assert!(matches!(
564            source.fetch(&client(), 8).await,
565            Err(Error::Subscription("subscription exceeds size limit"))
566        ));
567        server.await.unwrap();
568
569        let (source, server) = serve(
570            "HTTP/1.1 200 OK\r\nContent-Length: 17\r\nConnection: close\r\n\r\nhttp://host:8080\n",
571        )
572        .await;
573        assert_eq!(source.fetch(&client(), 17).await.unwrap().nodes.len(), 1);
574        server.await.unwrap();
575    }
576}
subscription_proxy_pool/subscription.rs

subscription_proxy_pool/
subscription.rs
