use crate::error::{ConfigError, ConfigResult};
use crate::loader::{detect_format_from_content, parse_content, Format};
use crate::types::{AnnotatedValue, SourceId};
use arc_swap::ArcSwap;
use async_trait::async_trait;
use reqwest::Client;
use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
use std::sync::atomic::AtomicU64;
use std::sync::Arc;
use std::sync::LazyLock;
use std::time::Duration;
use tokio::sync::RwLock;
pub const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(60);
static BLOCKED_NETWORKS: LazyLock<Vec<ipnet::IpNet>> = LazyLock::new(|| {
vec![
"127.0.0.0/8".parse().unwrap(),
"10.0.0.0/8".parse().unwrap(),
"172.16.0.0/12".parse().unwrap(),
"192.168.0.0/16".parse().unwrap(),
"169.254.0.0/16".parse().unwrap(),
"100.64.0.0/10".parse().unwrap(),
"192.0.2.0/24".parse().unwrap(),
"198.51.100.0/24".parse().unwrap(),
"203.0.113.0/24".parse().unwrap(),
"192.0.0.0/24".parse().unwrap(),
"fc00::/7".parse().unwrap(),
"fe80::/10".parse().unwrap(),
]
});
pub fn is_ip_blocked(ip: IpAddr) -> bool {
if let IpAddr::V6(ipv6) = ip {
let octets = ipv6.octets();
if octets[..10] == [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
&& octets[10] == 0xff
&& octets[11] == 0xff
{
return true;
}
}
if ip.is_loopback() {
return true;
}
BLOCKED_NETWORKS.iter().any(|net| net.contains(&ip))
}
fn resolve_host_with_validation(host: &str, port: u16) -> ConfigResult<Vec<IpAddr>> {
let addr_string = if port == 0 {
format!("{}:80", host)
} else {
format!("{}:{}", host, port)
};
let addrs: Vec<SocketAddr> = addr_string
.to_socket_addrs()
.map_err(|_| ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "resolvable hostname".to_string(),
message: format!("Cannot resolve hostname: {}", host),
})?
.collect();
if addrs.is_empty() {
return Err(ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "resolvable hostname".to_string(),
message: format!("No addresses resolved for hostname: {}", host),
});
}
let mut resolved_ips = Vec::new();
for addr in &addrs {
let ip = addr.ip();
resolved_ips.push(ip);
if is_ip_blocked(ip) {
return Err(ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "public IP".to_string(),
message: "SSRF attempt detected: resolved IP address is in a blocked private range"
.to_string(),
});
}
}
Ok(resolved_ips)
}
fn validate_url(url: &str, allowed_domains: &[String]) -> ConfigResult<Vec<IpAddr>> {
let parsed = url::Url::parse(url).map_err(|_| ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "valid URL".to_string(),
message: "Invalid URL format".to_string(),
})?;
if parsed.scheme() != "https" {
return Err(ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "https URL".to_string(),
message: "Only HTTPS URLs are allowed for remote configuration".to_string(),
});
}
let host = match parsed.host() {
Some(h) => h,
None => {
return Err(ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "valid URL with host".to_string(),
message: "URL must have a host".to_string(),
});
}
};
match host {
url::Host::Domain(domain) => {
let domain_str = domain.to_string();
let is_whitelisted = allowed_domains.iter().any(|allowed| {
if allowed == &domain_str {
return true;
}
if let Some(suffix) = allowed.strip_prefix("*.") {
return domain_str.ends_with(&format!(".{suffix}"))
&& !domain_str.ends_with(suffix);
}
domain_str.ends_with(&format!(".{allowed}"))
});
if is_whitelisted {
return Ok(Vec::new());
}
let port = parsed.port().unwrap_or(443);
let resolved = resolve_host_with_validation(&domain_str, port)?;
Ok(resolved)
}
url::Host::Ipv4(ip) => {
if is_ip_blocked(IpAddr::V4(ip)) {
return Err(ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "public IP".to_string(),
message: "Connection to private/internal IP addresses is not allowed"
.to_string(),
});
}
Ok(vec![IpAddr::V4(ip)])
}
url::Host::Ipv6(ip) => {
if is_ip_blocked(IpAddr::V6(ip)) {
return Err(ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "public IP".to_string(),
message: "Connection to private/internal IP addresses is not allowed"
.to_string(),
});
}
Ok(vec![IpAddr::V6(ip)])
}
}
}
#[async_trait]
pub trait PolledSource: Send + Sync {
async fn poll(&self) -> ConfigResult<AnnotatedValue>;
fn poll_interval(&self) -> Option<Duration>;
fn source_id(&self) -> SourceId;
}
#[derive(Debug)]
pub struct HttpPolledSource {
url: Arc<str>,
interval: Duration,
client: Client,
format: Option<Format>,
#[allow(dead_code)] cache_generation: AtomicU64,
cached: RwLock<Option<AnnotatedValue>>,
last_etag: ArcSwap<Option<String>>,
last_modified: ArcSwap<Option<String>>,
source_id: SourceId,
}
pub struct HttpPolledSourceBuilder {
url: Option<String>,
interval: Option<Duration>,
format: Option<Format>,
timeout: Option<Duration>,
allowed_domains: Vec<String>,
}
impl HttpPolledSourceBuilder {
pub fn new() -> Self {
Self {
url: None,
interval: None,
format: None,
timeout: None,
allowed_domains: Vec::new(),
}
}
pub fn url(mut self, url: impl Into<String>) -> Self {
self.url = Some(url.into());
self
}
pub fn interval(mut self, interval: Duration) -> Self {
self.interval = Some(interval);
self
}
pub fn format(mut self, format: Format) -> Self {
self.format = Some(format);
self
}
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = Some(timeout);
self
}
pub fn allowed_domain(mut self, domain: impl Into<String>) -> Self {
self.allowed_domains.push(domain.into());
self
}
pub fn allowed_domains(mut self, domains: impl IntoIterator<Item = impl Into<String>>) -> Self {
for domain in domains {
self.allowed_domains.push(domain.into());
}
self
}
pub fn build(self) -> ConfigResult<HttpPolledSource> {
let url = self.url.ok_or_else(|| ConfigError::InvalidValue {
key: "url".to_string(),
expected_type: "string".to_string(),
message: "URL is required".to_string(),
})?;
validate_url(&url, &self.allowed_domains)?;
let url_arc: Arc<str> = url.clone().into();
let source_id = SourceId::new(format!("http:{}", url_arc));
let mut client_builder = Client::builder().use_rustls_tls();
if let Some(timeout) = self.timeout {
client_builder = client_builder.timeout(timeout);
}
let client = client_builder
.build()
.map_err(|_e| ConfigError::RemoteUnavailable {
error_type: "ClientBuild".to_string(),
retryable: false,
})?;
Ok(HttpPolledSource {
url: url_arc,
interval: self.interval.unwrap_or(DEFAULT_POLL_INTERVAL),
client,
format: self.format,
cache_generation: AtomicU64::new(0),
cached: RwLock::new(None),
last_etag: ArcSwap::new(Arc::new(None)),
last_modified: ArcSwap::new(Arc::new(None)),
source_id,
})
}
}
impl Default for HttpPolledSourceBuilder {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl PolledSource for HttpPolledSource {
async fn poll(&self) -> ConfigResult<AnnotatedValue> {
let mut request = self.client.get(self.url.as_ref());
if let Some(etag) = self.last_etag.load().as_ref() {
request = request.header("If-None-Match", etag.as_str());
}
if let Some(modified) = self.last_modified.load().as_ref() {
request = request.header("If-Modified-Since", modified.as_str());
}
let response = request
.send()
.await
.map_err(|e| ConfigError::RemoteUnavailable {
error_type: std::any::type_name::<reqwest::Error>().to_string(),
retryable: is_retryable_error(&e),
})?;
let status = response.status();
if status == reqwest::StatusCode::NOT_MODIFIED {
if let Some(cached) = self.cached.read().await.as_ref() {
return Ok(cached.clone());
}
return Err(ConfigError::RemoteUnavailable {
error_type: "NoCachedValue".to_string(),
retryable: false,
});
}
if !status.is_success() {
return Err(ConfigError::RemoteUnavailable {
error_type: format!("HTTP_{}", status.as_u16()),
retryable: status.is_server_error() || status.as_u16() == 429,
});
}
if let Some(etag) = response.headers().get("etag") {
if let Ok(etag_str) = etag.to_str() {
self.last_etag.store(Arc::new(Some(etag_str.to_string())));
}
}
if let Some(modified) = response.headers().get("last-modified") {
if let Ok(modified_str) = modified.to_str() {
self.last_modified
.store(Arc::new(Some(modified_str.to_string())));
}
}
let body = response
.text()
.await
.map_err(|e| ConfigError::RemoteUnavailable {
error_type: std::any::type_name::<reqwest::Error>().to_string(),
retryable: is_retryable_error(&e),
})?;
let format = self
.format
.unwrap_or_else(|| detect_format_from_content(&body).unwrap_or(Format::Json));
let source = self.source_id.clone();
let value = parse_remote_content(&body, format, source)?;
*self.cached.write().await = Some(value.clone());
Ok(value)
}
fn poll_interval(&self) -> Option<Duration> {
Some(self.interval)
}
fn source_id(&self) -> SourceId {
self.source_id.clone()
}
}
fn parse_remote_content(
content: &str,
format: Format,
source: SourceId,
) -> ConfigResult<AnnotatedValue> {
parse_content(content, format, source, None)
}
fn is_retryable_error(error: &reqwest::Error) -> bool {
error.is_timeout() || error.is_connect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_is_ip_blocked_loopback_v4() {
assert!(is_ip_blocked(IpAddr::V4("127.0.0.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4(
"127.255.255.255".parse().unwrap()
)));
assert!(is_ip_blocked(IpAddr::V4("127.0.0.0".parse().unwrap())));
}
#[test]
fn test_is_ip_blocked_private_v4() {
assert!(is_ip_blocked(IpAddr::V4("10.0.0.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("10.255.255.255".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("172.16.0.0".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("172.31.255.255".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("172.20.0.1".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V4("172.15.0.1".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V4("172.32.0.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("192.168.0.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4(
"192.168.255.255".parse().unwrap()
)));
}
#[test]
fn test_is_ip_blocked_link_local_v4() {
assert!(is_ip_blocked(IpAddr::V4("169.254.0.0".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4(
"169.254.255.255".parse().unwrap()
)));
}
#[test]
fn test_is_ip_blocked_carrier_nat_v4() {
assert!(is_ip_blocked(IpAddr::V4("100.64.0.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4(
"100.127.255.255".parse().unwrap()
)));
assert!(!is_ip_blocked(IpAddr::V4("100.0.0.1".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V4("100.128.0.1".parse().unwrap())));
}
#[test]
fn test_is_ip_blocked_documentation_v4() {
assert!(is_ip_blocked(IpAddr::V4("192.0.2.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("198.51.100.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("203.0.113.1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V4("192.0.0.1".parse().unwrap())));
}
#[test]
fn test_is_ip_blocked_public_v4() {
assert!(!is_ip_blocked(IpAddr::V4("8.8.8.8".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V4("1.1.1.1".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V4("93.184.216.34".parse().unwrap()))); assert!(!is_ip_blocked(IpAddr::V4("52.94.236.248".parse().unwrap()))); }
#[test]
fn test_is_ip_blocked_loopback_v6() {
assert!(is_ip_blocked(IpAddr::V6("::1".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V6("::0".parse().unwrap())));
}
#[test]
fn test_is_ip_blocked_unique_local_v6() {
assert!(is_ip_blocked(IpAddr::V6("fc00::1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V6("fd00::1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V6("fdff::1".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V6("fe00::1".parse().unwrap())));
}
#[test]
fn test_is_ip_blocked_link_local_v6() {
assert!(is_ip_blocked(IpAddr::V6("fe80::1".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V6(
"fe80:ffff:ffff:ffff::".parse().unwrap()
)));
assert!(is_ip_blocked(IpAddr::V6("fe81::1".parse().unwrap())));
assert!(!is_ip_blocked(IpAddr::V6("fe7f::1".parse().unwrap())));
}
#[test]
fn test_is_ip_blocked_ipv4_mapped_v6() {
assert!(is_ip_blocked(IpAddr::V6(
"::ffff:127.0.0.1".parse().unwrap()
)));
assert!(is_ip_blocked(IpAddr::V6("::ffff:0:0".parse().unwrap())));
assert!(is_ip_blocked(IpAddr::V6("::ffff:8.8.8.8".parse().unwrap())));
}
#[test]
fn test_is_ip_blocked_public_v6() {
assert!(!is_ip_blocked(IpAddr::V6(
"2001:4860:4860::8888".parse().unwrap()
))); assert!(!is_ip_blocked(IpAddr::V6(
"2606:4700:4700::1111".parse().unwrap()
))); }
#[test]
fn test_validate_url_rejects_non_https() {
let result = validate_url("http://example.com/config.json", &[]);
assert!(result.is_err());
let err = result.unwrap_err();
assert!(matches!(err, ConfigError::InvalidValue { .. }));
}
#[test]
fn test_validate_url_rejects_private_ipv4() {
let result = validate_url("https://127.0.0.1/config.json", &[]);
assert!(result.is_err());
let result = validate_url("https://10.0.0.1/config.json", &[]);
assert!(result.is_err());
let result = validate_url("https://192.168.1.1/config.json", &[]);
assert!(err_if_blocked(&result));
assert!(result.is_err());
let result = validate_url("https://172.16.0.1/config.json", &[]);
assert!(result.is_err());
}
#[test]
fn test_validate_url_rejects_private_ipv6() {
let result = validate_url("https://[::1]/config.json", &[]);
assert!(result.is_err());
let result = validate_url("https://[fe80::1]/config.json", &[]);
assert!(result.is_err());
let result = validate_url("https://[fc00::1]/config.json", &[]);
assert!(result.is_err());
}
#[test]
fn test_validate_url_rejects_ipv4_mapped() {
let result = validate_url("https://[::ffff:127.0.0.1]/config.json", &[]);
assert!(result.is_err());
}
#[test]
fn test_validate_url_rejects_documentation_ips() {
let result = validate_url("https://192.0.2.1/config.json", &[]);
assert!(result.is_err());
let result = validate_url("https://198.51.100.1/config.json", &[]);
assert!(result.is_err());
let result = validate_url("https://203.0.113.1/config.json", &[]);
assert!(result.is_err());
let result = validate_url("https://192.0.0.1/config.json", &[]);
assert!(result.is_err());
}
#[test]
fn test_validate_url_accepts_public_ips() {
let _result = validate_url("https://8.8.8.8/config.json", &[]);
}
#[test]
fn test_validate_url_whitelist_exact_match() {
let result = validate_url(
"https://internal.example.com/config.json",
&["internal.example.com".to_string()],
);
assert!(
result.is_ok(),
"whitelisted domain should be accepted: {:?}",
result.err()
);
}
#[test]
fn test_validate_url_whitelist_subdomain_match() {
let result = validate_url(
"https://api.example.com/config.json",
&["example.com".to_string()],
);
assert!(
result.is_ok(),
"subdomain of whitelisted domain should be accepted: {:?}",
result.err()
);
}
#[test]
fn test_validate_url_whitelist_no_match() {
let result = validate_url(
"https://untrusted.example.com/config.json",
&["trusted.example.com".to_string()],
);
assert!(
result.is_err(),
"non-whitelisted domain should be rejected: {result:?}"
);
}
#[test]
fn test_validate_url_whitelist_mixed() {
let domains = vec![
"internal.corp.com".to_string(),
"config-service.prod".to_string(),
];
let result = validate_url("https://internal.corp.com/config.json", &domains);
assert!(result.is_ok());
let result = validate_url("https://config-service.prod/config.json", &domains);
assert!(result.is_ok());
}
fn err_if_blocked(result: &Result<Vec<IpAddr>, ConfigError>) -> bool {
if let Err(e) = result {
matches!(e, ConfigError::InvalidValue { .. })
} else {
false
}
}
#[test]
fn test_http_polled_source_builder() {
let source = HttpPolledSourceBuilder::new()
.url("https://example.com/config.json")
.interval(Duration::from_secs(30))
.build()
.expect("builder should succeed with valid public URL");
assert_eq!(source.poll_interval(), Some(Duration::from_secs(30)));
assert_eq!(
source.source_id().as_str(),
"http:https://example.com/config.json"
);
}
#[test]
fn test_http_polled_source_builder_requires_url() {
let result = HttpPolledSourceBuilder::new().build();
assert!(result.is_err());
}
#[test]
fn test_default_poll_interval() {
let source = HttpPolledSourceBuilder::new()
.url("https://example.com/config.json")
.build()
.expect("builder should succeed with valid public URL");
assert_eq!(source.poll_interval(), Some(DEFAULT_POLL_INTERVAL));
}
#[test]
fn test_builder_pattern() {
let source = HttpPolledSourceBuilder::new()
.url("https://example.com/config.yaml")
.interval(Duration::from_secs(30))
.format(Format::Yaml)
.timeout(Duration::from_secs(5))
.allowed_domain("example.com")
.allowed_domains(["cdn.example.com", "assets.example.com"])
.build()
.expect("builder should succeed with valid public URL and whitelist");
assert_eq!(source.poll_interval(), Some(Duration::from_secs(30)));
}
#[test]
fn test_builder_rejects_blocked_ip() {
let result = HttpPolledSourceBuilder::new()
.url("https://192.168.1.1/config.json")
.build();
assert!(result.is_err());
let err = result.unwrap_err();
let msg = err.user_message();
assert!(
msg.contains("private") || msg.contains("SSRF") || msg.contains("blocked"),
"Expected SSRF-related error message, got: {}",
msg
);
}
#[test]
fn test_builder_rejects_loopback() {
let result = HttpPolledSourceBuilder::new()
.url("https://127.0.0.1/config.json")
.build();
assert!(result.is_err());
let result = HttpPolledSourceBuilder::new()
.url("https://[::1]/config.json")
.build();
assert!(result.is_err());
}
#[test]
fn test_builder_accepts_whitelisted_domain() {
let result = HttpPolledSourceBuilder::new()
.url("https://whitelisted-internal.local/config.json")
.allowed_domain("whitelisted-internal.local")
.build();
match result {
Ok(_) => {}
Err(ConfigError::InvalidValue { message, .. }) => {
assert!(
message.contains("resolve") || message.contains("Cannot resolve"),
"Expected DNS resolution error, got: {}",
message
);
}
Err(_) => {}
}
}
#[test]
fn test_resolve_host_with_validation_public() {
let result = resolve_host_with_validation("example.com", 443);
if let Ok(ips) = result {
assert!(!ips.is_empty());
for ip in &ips {
assert!(
!is_ip_blocked(*ip),
"example.com resolved to a blocked IP: {}",
ip
);
}
}
}
}