1use async_trait::async_trait;
15use uuid::Uuid;
16
17use crate::error::ProxyResult;
18use crate::types::{Proxy, ProxyRecord};
19
20#[async_trait]
48pub trait ProxyStoragePort: Send + Sync + 'static {
49 async fn add(&self, proxy: Proxy) -> ProxyResult<ProxyRecord>;
51
52 async fn remove(&self, id: Uuid) -> ProxyResult<()>;
54
55 async fn list(&self) -> ProxyResult<Vec<ProxyRecord>>;
57
58 async fn get(&self, id: Uuid) -> ProxyResult<ProxyRecord>;
60
61 async fn update_metrics(&self, id: Uuid, success: bool, latency_ms: u64) -> ProxyResult<()>;
66
67 async fn list_with_metrics(&self) -> ProxyResult<Vec<(ProxyRecord, Arc<ProxyMetrics>)>>;
73}
74
75pub type BoxedProxyStorage = Box<dyn ProxyStoragePort>;
77
78fn validate_proxy_url(url: &str) -> ProxyResult<()> {
91 use crate::error::ProxyError;
92 use crate::vendor_quirks::{self, ParseError, QuirkSeverity};
93
94 let parsed = vendor_quirks::ProxyUrl::parse(url).map_err(|e| match e {
102 ParseError::MissingSchemeSeparator(_) => ProxyError::InvalidProxyUrl {
103 url: url.to_owned(),
104 reason: "missing scheme separator '://'".into(),
105 },
106 ParseError::UnsupportedScheme(ref s, _) => ProxyError::InvalidProxyUrl {
107 url: url.to_owned(),
108 reason: format!("unsupported scheme '{s}'"),
109 },
110 ParseError::EmptyHost(_) => ProxyError::InvalidProxyUrl {
111 url: url.to_owned(),
112 reason: "empty host".into(),
113 },
114 ParseError::NonNumericPort(ref p, _) => ProxyError::InvalidProxyUrl {
115 url: url.to_owned(),
116 reason: format!("non-numeric port '{p}'"),
117 },
118 ParseError::PortOutOfRange { ref port, .. } => ProxyError::InvalidProxyUrl {
119 url: url.to_owned(),
120 reason: format!("port {port} is out of range [1, 65535]"),
121 },
122 ParseError::UnclosedIpv6Bracket(_) => ProxyError::InvalidProxyUrl {
123 url: url.to_owned(),
124 reason: "unclosed IPv6 bracket in host".into(),
125 },
126 })?;
127
128 let quirks = vendor_quirks::check(&parsed);
142 for m in &quirks {
143 match m.severity {
144 QuirkSeverity::Error => {
145 return Err(ProxyError::InvalidProxyUrl {
146 url: url.to_owned(),
147 reason: m.description.to_owned(),
148 });
149 }
150 QuirkSeverity::Warning => {
151 tracing::warn!(
152 proxy_url_host = %parsed.host,
153 proxy_url_port = ?parsed.port,
154 quirk_host_suffix = m.host_suffix,
155 observed_scheme = m.observed_scheme.as_str(),
156 required_scheme = m.required_scheme.as_str(),
157 quirk_description = m.description,
158 "proxy URL matches a vendor quirk warning (URL accepted)"
159 );
160 }
161 QuirkSeverity::Info => {
162 tracing::info!(
163 proxy_url_host = %parsed.host,
164 proxy_url_port = ?parsed.port,
165 quirk_host_suffix = m.host_suffix,
166 quirk_description = m.description,
167 "proxy URL matches a vendor quirk info record"
168 );
169 }
170 }
171 }
172
173 Ok(())
174}
175
176fn validate_geo_metadata(caps: &crate::types::ProxyCapabilities) -> ProxyResult<()> {
190 use crate::types::{validate_asn, validate_city, validate_postal_code};
191
192 if let Some(asn) = caps.asn {
193 validate_asn(asn)?;
194 }
195 if let Some(ref city) = caps.city {
196 validate_city(city)?;
197 }
198 if let Some(ref postal) = caps.postal_code {
199 validate_postal_code(postal)?;
200 }
201 Ok(())
202}
203
204use std::collections::HashMap;
209use tokio::sync::RwLock;
210
211use crate::types::ProxyMetrics;
212use std::sync::Arc;
213
214type StoreMap = HashMap<Uuid, (ProxyRecord, Arc<ProxyMetrics>)>;
215
216#[derive(Debug, Default, Clone)]
242pub struct MemoryProxyStore {
243 inner: Arc<RwLock<StoreMap>>,
244}
245
246impl MemoryProxyStore {
247 pub async fn with_proxies(proxies: Vec<Proxy>) -> ProxyResult<Self> {
257 let store = Self::default();
258 for proxy in proxies {
259 store.add(proxy).await?;
260 }
261 Ok(store)
262 }
263}
264
265#[async_trait]
266impl ProxyStoragePort for MemoryProxyStore {
267 async fn add(&self, proxy: Proxy) -> ProxyResult<ProxyRecord> {
268 validate_proxy_url(&proxy.url)?;
269 validate_geo_metadata(&proxy.capabilities)?;
270 let record = ProxyRecord::new(proxy);
271 let metrics = Arc::new(ProxyMetrics::default());
272 self.inner
273 .write()
274 .await
275 .insert(record.id, (record.clone(), metrics));
276 Ok(record)
277 }
278
279 async fn remove(&self, id: Uuid) -> ProxyResult<()> {
280 self.inner
281 .write()
282 .await
283 .remove(&id)
284 .map(|_| ())
285 .ok_or_else(|| crate::error::ProxyError::StorageError(format!("proxy {id} not found")))
286 }
287
288 async fn list(&self) -> ProxyResult<Vec<ProxyRecord>> {
289 Ok(self
290 .inner
291 .read()
292 .await
293 .values()
294 .map(|(r, _)| r.clone())
295 .collect())
296 }
297
298 async fn get(&self, id: Uuid) -> ProxyResult<ProxyRecord> {
299 self.inner
300 .read()
301 .await
302 .get(&id)
303 .map(|(r, _)| r.clone())
304 .ok_or_else(|| crate::error::ProxyError::StorageError(format!("proxy {id} not found")))
305 }
306
307 async fn list_with_metrics(&self) -> ProxyResult<Vec<(ProxyRecord, Arc<ProxyMetrics>)>> {
308 Ok(self
309 .inner
310 .read()
311 .await
312 .values()
313 .map(|(r, m)| (r.clone(), Arc::clone(m)))
314 .collect())
315 }
316
317 async fn update_metrics(&self, id: Uuid, success: bool, latency_ms: u64) -> ProxyResult<()> {
318 use std::sync::atomic::Ordering;
319
320 let metrics = self
321 .inner
322 .read()
323 .await
324 .get(&id)
325 .map(|(_, m)| Arc::clone(m))
326 .ok_or_else(|| {
327 crate::error::ProxyError::StorageError(format!("proxy {id} not found"))
328 })?;
329
330 metrics.requests_total.fetch_add(1, Ordering::Relaxed);
332 if success {
333 metrics.successes.fetch_add(1, Ordering::Relaxed);
334 } else {
335 metrics.failures.fetch_add(1, Ordering::Relaxed);
336 }
337 metrics
338 .total_latency_ms
339 .fetch_add(latency_ms, Ordering::Relaxed);
340 Ok(())
341 }
342}
343
344#[cfg(test)]
349#[allow(
350 clippy::unwrap_used,
351 clippy::expect_used,
352 clippy::panic,
353 clippy::indexing_slicing
354)] mod tests {
356 use super::*;
357 use crate::types::ProxyType;
358 use std::sync::atomic::Ordering;
359
360 fn make_proxy(url: &str) -> Proxy {
361 Proxy {
362 url: url.into(),
363 proxy_type: ProxyType::Http,
364 username: None,
365 password: None,
366 weight: 1,
367 tags: vec![],
368 capabilities: crate::types::ProxyCapabilities::default(),
369 ip_class: crate::types::IpClass::Unknown,
370 target_compatibility: crate::types::TargetVendorCompatibility::default(),
371 }
372 }
373
374 #[tokio::test]
375 async fn add_list_remove() -> crate::error::ProxyResult<()> {
376 let store = MemoryProxyStore::default();
377 let r1 = store.add(make_proxy("http://a.test:8080")).await?;
378 let r2 = store.add(make_proxy("http://b.test:8080")).await?;
379 let r3 = store.add(make_proxy("http://c.test:8080")).await?;
380 assert_eq!(store.list().await?.len(), 3);
381 store.remove(r2.id).await?;
382 let remaining = store.list().await?;
383 assert_eq!(remaining.len(), 2);
384 let ids: Vec<_> = remaining.iter().map(|r| r.id).collect();
385 assert!(ids.contains(&r1.id));
386 assert!(ids.contains(&r3.id));
387 Ok(())
388 }
389
390 #[tokio::test]
391 async fn invalid_url_rejected() -> std::result::Result<(), Box<dyn std::error::Error>> {
392 let store = MemoryProxyStore::default();
393 let err = store
394 .add(make_proxy("not-a-url"))
395 .await
396 .err()
397 .ok_or_else(|| std::io::Error::other("invalid URL should be rejected"))?;
398 assert!(matches!(
399 err,
400 crate::error::ProxyError::InvalidProxyUrl { .. }
401 ));
402 Ok(())
403 }
404
405 #[tokio::test]
406 async fn invalid_url_empty_host() -> std::result::Result<(), Box<dyn std::error::Error>> {
407 let store = MemoryProxyStore::default();
408 let err = store
409 .add(make_proxy("http://:8080"))
410 .await
411 .err()
412 .ok_or_else(|| std::io::Error::other("empty host URL should be rejected"))?;
413 assert!(matches!(
414 err,
415 crate::error::ProxyError::InvalidProxyUrl { .. }
416 ));
417 Ok(())
418 }
419
420 #[tokio::test]
424 async fn invalid_geo_metadata_asn_zero_rejected()
425 -> std::result::Result<(), Box<dyn std::error::Error>> {
426 let store = MemoryProxyStore::default();
427 let mut p = make_proxy("http://cf.test:8080");
428 p.capabilities.asn = Some(0);
429 let err = store
430 .add(p)
431 .await
432 .err()
433 .ok_or_else(|| std::io::Error::other("asn=0 should be rejected"))?;
434 assert!(matches!(
435 err,
436 crate::error::ProxyError::InvalidGeoMetadata { ref field, .. } if field == "asn"
437 ));
438 Ok(())
439 }
440
441 #[tokio::test]
443 async fn invalid_geo_metadata_asn_max_rejected()
444 -> std::result::Result<(), Box<dyn std::error::Error>> {
445 let store = MemoryProxyStore::default();
446 let mut p = make_proxy("http://cf.test:8080");
447 p.capabilities.asn = Some(u32::MAX);
448 let err = store
449 .add(p)
450 .await
451 .err()
452 .ok_or_else(|| std::io::Error::other("asn=u32::MAX should be rejected"))?;
453 assert!(matches!(
454 err,
455 crate::error::ProxyError::InvalidGeoMetadata { ref field, .. } if field == "asn"
456 ));
457 Ok(())
458 }
459
460 #[tokio::test]
462 async fn invalid_geo_metadata_empty_city_rejected()
463 -> std::result::Result<(), Box<dyn std::error::Error>> {
464 let store = MemoryProxyStore::default();
465 let mut p = make_proxy("http://cf.test:8080");
466 p.capabilities.city = Some(String::new());
467 let err = store
468 .add(p)
469 .await
470 .err()
471 .ok_or_else(|| std::io::Error::other("empty city should be rejected"))?;
472 assert!(matches!(
473 err,
474 crate::error::ProxyError::InvalidGeoMetadata { ref field, .. } if field == "city"
475 ));
476 Ok(())
477 }
478
479 #[tokio::test]
481 async fn invalid_geo_metadata_empty_postal_code_rejected()
482 -> std::result::Result<(), Box<dyn std::error::Error>> {
483 let store = MemoryProxyStore::default();
484 let mut p = make_proxy("http://cf.test:8080");
485 p.capabilities.postal_code = Some(String::new());
486 let err = store
487 .add(p)
488 .await
489 .err()
490 .ok_or_else(|| std::io::Error::other("empty postal_code should be rejected"))?;
491 assert!(matches!(
492 err,
493 crate::error::ProxyError::InvalidGeoMetadata { ref field, .. } if field == "postal_code"
494 ));
495 Ok(())
496 }
497
498 #[tokio::test]
500 async fn valid_geo_metadata_accepted() -> std::result::Result<(), Box<dyn std::error::Error>> {
501 let store = MemoryProxyStore::default();
502 let mut p = make_proxy("http://cf.test:8080");
503 p.capabilities.asn = Some(13_335);
504 p.capabilities.city = Some("San Francisco".into());
505 p.capabilities.postal_code = Some("94110".into());
506 let record = store
507 .add(p)
508 .await
509 .map_err(|e| std::io::Error::other(format!("expected accept, got {e}")))?;
510 assert_eq!(record.proxy.capabilities.asn, Some(13_335));
511 Ok(())
512 }
513
514 #[tokio::test]
515 async fn concurrent_metrics_updates() -> std::result::Result<(), Box<dyn std::error::Error>> {
516 use tokio::task::JoinSet;
517
518 let store = Arc::new(MemoryProxyStore::default());
519 let record = store
520 .add(make_proxy("http://proxy.test:3128"))
521 .await
522 .map_err(|e| std::io::Error::other(format!("failed to add proxy: {e}")))?;
523 let id = record.id;
524
525 let mut tasks = JoinSet::new();
526 for i in 0u64..50 {
527 let s = Arc::clone(&store);
528 tasks.spawn(async move { s.update_metrics(id, i % 2 == 0, i * 10).await });
529 }
530 while let Some(res) = tasks.join_next().await {
531 let inner = res.map_err(|e| std::io::Error::other(format!("join failed: {e}")))?;
532 inner.map_err(|e| std::io::Error::other(format!("update_metrics failed: {e}")))?;
533 }
534
535 let guard = store.inner.read().await;
537 let metrics = guard
538 .get(&id)
539 .map(|(_, m)| Arc::clone(m))
540 .ok_or_else(|| std::io::Error::other("missing metrics for inserted proxy"))?;
541 drop(guard);
542
543 let total = metrics.requests_total.load(Ordering::Relaxed);
544 let successes = metrics.successes.load(Ordering::Relaxed);
545 let failures = metrics.failures.load(Ordering::Relaxed);
546 assert_eq!(total, 50);
547 assert_eq!(successes + failures, 50);
548 Ok(())
549 }
550
551 #[tokio::test]
557 async fn validate_crawlera_https_8011_rejected()
558 -> std::result::Result<(), Box<dyn std::error::Error>> {
559 let store = MemoryProxyStore::default();
560 let err = store
561 .add(make_proxy("https://user:secret@proxy.crawlera.com:8011"))
562 .await
563 .err()
564 .ok_or_else(|| std::io::Error::other("Crawlera 8011 + https:// must be rejected"))?;
565 match err {
566 crate::error::ProxyError::InvalidProxyUrl { url, reason } => {
567 assert_eq!(url, "https://user:secret@proxy.crawlera.com:8011");
568 assert!(
570 !reason.contains("secret"),
571 "reason leaked password: {reason}"
572 );
573 assert!(
574 !reason.contains("user:secret"),
575 "reason leaked credentials: {reason}"
576 );
577 assert!(
579 reason.contains("WRONG_VERSION_NUMBER") || reason.contains("plain HTTP"),
580 "reason should reference the quirk, got: {reason}"
581 );
582 }
583 other => panic!("expected InvalidProxyUrl, got {other:?}"),
584 }
585 Ok(())
586 }
587
588 #[tokio::test]
591 async fn validate_crawlera_http_8011_accepted()
592 -> std::result::Result<(), Box<dyn std::error::Error>> {
593 let store = MemoryProxyStore::default();
594 let record = store
595 .add(make_proxy("http://apikey:@proxy.crawlera.com:8011"))
596 .await
597 .map_err(|e| {
598 std::io::Error::other(format!("Crawlera 8011 + http:// must be accepted: {e}"))
599 })?;
600 assert_eq!(record.proxy.url, "http://apikey:@proxy.crawlera.com:8011");
601 Ok(())
602 }
603
604 #[tokio::test]
607 async fn validate_zyte_https_8011_rejected()
608 -> std::result::Result<(), Box<dyn std::error::Error>> {
609 let store = MemoryProxyStore::default();
610 let err = store
611 .add(make_proxy("https://apikey:@proxy.zyte.com:8011"))
612 .await
613 .err()
614 .ok_or_else(|| std::io::Error::other("Zyte 8011 + https:// must be rejected"))?;
615 assert!(matches!(
616 err,
617 crate::error::ProxyError::InvalidProxyUrl { ref reason, .. }
618 if reason.contains("WRONG_VERSION_NUMBER") || reason.contains("plain HTTP")
619 ));
620 Ok(())
621 }
622
623 #[tokio::test]
626 async fn validate_bright_data_warning_accepted()
627 -> std::result::Result<(), Box<dyn std::error::Error>> {
628 let store = MemoryProxyStore::default();
629 let record = store
630 .add(make_proxy(
631 "http://brd-customer-1-session-abc123@brd.superproxy.io:22225",
632 ))
633 .await
634 .map_err(|e| {
635 std::io::Error::other(format!("Bright Data Warning URL must be accepted: {e}"))
636 })?;
637 assert!(record.proxy.url.contains("brd.superproxy.io"));
638 Ok(())
639 }
640
641 #[tokio::test]
643 async fn validate_iproyal_warning_accepted()
644 -> std::result::Result<(), Box<dyn std::error::Error>> {
645 let store = MemoryProxyStore::default();
646 let record = store
647 .add(make_proxy(
648 "http://user-country-US:pass@residential.iproyal.com:12321",
649 ))
650 .await
651 .map_err(|e| {
652 std::io::Error::other(format!("IPRoyal Warning URL must be accepted: {e}"))
653 })?;
654 assert!(record.proxy.url.contains("iproyal.com"));
655 Ok(())
656 }
657
658 #[tokio::test]
661 async fn validate_unknown_host_no_quirks_accepted()
662 -> std::result::Result<(), Box<dyn std::error::Error>> {
663 let store = MemoryProxyStore::default();
664 let record = store
665 .add(make_proxy(
666 "http://user:pass@some-unrelated-host.example:8080",
667 ))
668 .await
669 .map_err(|e| std::io::Error::other(format!("unrelated host must be accepted: {e}")))?;
670 assert!(record.proxy.url.contains("some-unrelated-host.example"));
671 Ok(())
672 }
673
674 #[tokio::test]
677 async fn validate_preserves_structural_rejection()
678 -> std::result::Result<(), Box<dyn std::error::Error>> {
679 let store = MemoryProxyStore::default();
680 let err = store
681 .add(make_proxy("not-a-url"))
682 .await
683 .err()
684 .ok_or_else(|| std::io::Error::other("not-a-url must be rejected"))?;
685 assert!(matches!(
686 err,
687 crate::error::ProxyError::InvalidProxyUrl { ref reason, .. }
688 if reason.contains("missing scheme separator")
689 ));
690 Ok(())
691 }
692
693 #[tokio::test]
696 async fn validate_preserves_empty_host_rejection()
697 -> std::result::Result<(), Box<dyn std::error::Error>> {
698 let store = MemoryProxyStore::default();
699 let err = store
700 .add(make_proxy("http://:8080"))
701 .await
702 .err()
703 .ok_or_else(|| std::io::Error::other("empty host must be rejected"))?;
704 assert!(matches!(
705 err,
706 crate::error::ProxyError::InvalidProxyUrl { ref reason, .. }
707 if reason.contains("empty host")
708 ));
709 Ok(())
710 }
711
712 #[tokio::test]
715 async fn validate_crawlera_non_8011_https_accepted()
716 -> std::result::Result<(), Box<dyn std::error::Error>> {
717 let store = MemoryProxyStore::default();
718 let record = store
719 .add(make_proxy("https://user:pass@proxy.crawlera.com:9000"))
720 .await
721 .map_err(|e| {
722 std::io::Error::other(format!("Crawlera 9000 + https:// must be accepted: {e}"))
723 })?;
724 assert!(record.proxy.url.contains("crawlera.com:9000"));
725 Ok(())
726 }
727}