//! Plugin sandboxing: token-bucket rate limiting, storage quotas, URL allowlists.

use bext_plugin_api::types::WasmPluginPermissions;
use std::sync::Mutex;
use std::time::Instant;

/// Per-plugin sandbox enforcing resource limits.
pub struct PluginSandbox {
    pub plugin_id: String,
    pub permissions: WasmPluginPermissions,
    /// Storage directory for this plugin (scoped by plugin_id).
    pub storage_dir: std::path::PathBuf,
    /// Token bucket for rate-limiting HTTP fetches.
    fetch_limiter: Mutex<TokenBucket>,
    /// Tracked storage usage in bytes.
    storage_bytes: Mutex<u64>,
}

impl PluginSandbox {
    pub fn new(
        plugin_id: String,
        permissions: WasmPluginPermissions,
        storage_root: &std::path::Path,
    ) -> Self {
        let storage_dir = storage_root.join(&plugin_id);
        let bucket = TokenBucket::new(
            permissions.max_fetch_per_minute,
            std::time::Duration::from_secs(60),
        );
        Self {
            plugin_id,
            permissions,
            storage_dir,
            fetch_limiter: Mutex::new(bucket),
            storage_bytes: Mutex::new(0),
        }
    }

    /// Check if an HTTP fetch URL is allowed by the permission globs.
    pub fn is_url_allowed(&self, url: &str) -> bool {
        if self.permissions.allowed_urls.is_empty() {
            return false; // No URLs allowed by default
        }
        self.permissions
            .allowed_urls
            .iter()
            .any(|pattern| glob_match(pattern, url))
    }

    /// Try to acquire a fetch token. Returns false if rate limit exceeded.
    pub fn try_fetch(&self) -> bool {
        self.fetch_limiter
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .try_acquire()
    }

    /// Check if writing `bytes` would exceed the storage quota.
    pub fn check_storage_quota(&self, additional_bytes: u64) -> bool {
        let current = self.storage_bytes.lock().unwrap_or_else(|e| e.into_inner());
        *current + additional_bytes <= self.permissions.storage_quota_kb * 1024
    }

    /// Record storage usage.
    pub fn record_storage(&self, bytes: u64) {
        let mut current = self.storage_bytes.lock().unwrap_or_else(|e| e.into_inner());
        *current += bytes;
    }

    /// Sanitize a storage key to prevent path traversal.
    pub fn sanitize_key(key: &str) -> Option<String> {
        if key.contains("..") || key.contains('/') || key.contains('\\') || key.contains('\0') {
            None
        } else {
            Some(key.to_string())
        }
    }

    /// Check if a KV namespace is allowed by this plugin's permissions.
    pub fn is_kv_namespace_allowed(&self, namespace: &str) -> bool {
        self.permissions
            .kv_namespaces
            .iter()
            .any(|n| n == namespace)
    }

    /// Check if a queue name is allowed by this plugin's permissions.
    pub fn is_queue_allowed(&self, queue_name: &str) -> bool {
        self.permissions.queue_names.iter().any(|n| n == queue_name)
    }
}

/// Simple glob matcher supporting `*` wildcards.
fn glob_match(pattern: &str, input: &str) -> bool {
    let parts: Vec<&str> = pattern.split('*').collect();
    if parts.len() == 1 {
        return pattern == input;
    }

    let mut pos = 0;
    // First part must be a prefix
    if !parts[0].is_empty() {
        if !input.starts_with(parts[0]) {
            return false;
        }
        pos = parts[0].len();
    }

    // Middle parts must appear in order
    for part in &parts[1..parts.len() - 1] {
        if part.is_empty() {
            continue;
        }
        match input[pos..].find(part) {
            Some(idx) => pos += idx + part.len(),
            None => return false,
        }
    }

    // Last part must be a suffix
    let last = parts[parts.len() - 1];
    if !last.is_empty() {
        input[pos..].ends_with(last)
    } else {
        true
    }
}

/// Check if a URL targets a private/internal IP address (SSRF prevention).
///
/// Returns `true` if the URL resolves to a private, loopback, or link-local
/// address that should NOT be accessible from plugin HTTP fetch.
pub fn is_private_url(url_str: &str) -> bool {
    resolve_and_pin_url(url_str).is_err()
}

/// Resolve a URL's hostname and validate that it does not target a private IP.
///
/// On success, returns a **pinned URL** where the hostname has been replaced
/// with the resolved IP address. This eliminates the DNS rebinding TOCTOU gap:
/// DNS is resolved exactly once, validated, and the IP is baked into the URL
/// so the HTTP client connects directly without re-resolving.
///
/// The original `Host` header value is returned separately so the caller can
/// set it on the outgoing request (servers often need the hostname, not the IP).
///
/// Returns `Err` if the URL is unparseable, resolves to a private IP, or DNS
/// fails.
pub fn resolve_and_pin_url(url_str: &str) -> Result<(String, String), String> {
    let parsed = url::Url::parse(url_str)
        .map_err(|e| format!("unparseable URL: {e}"))?;

    let host = parsed.host_str()
        .ok_or_else(|| "URL has no host".to_string())?;

    // Block "localhost" hostname variants
    let host_lower = host.to_lowercase();
    if host_lower == "localhost" || host_lower.ends_with(".localhost") {
        return Err("blocked: localhost".into());
    }

    let original_host = match parsed.port() {
        Some(p) => format!("{host}:{p}"),
        None => host.to_string(),
    };

    // If the host is already an IP literal, validate and return as-is.
    let ip_direct = host.parse::<std::net::IpAddr>()
        .or_else(|_| host.trim_start_matches('[').trim_end_matches(']').parse::<std::net::IpAddr>());
    if let Ok(ip) = ip_direct {
        if is_private_ip(ip) {
            return Err(format!("blocked: private IP {ip}"));
        }
        return Ok((url_str.to_string(), original_host));
    }

    // DNS resolution — resolve hostname and check ALL returned IPs.
    let port = parsed.port_or_known_default().unwrap_or(80);
    let addrs: Vec<_> = std::net::ToSocketAddrs::to_socket_addrs(&(host, port))
        .map_err(|e| format!("DNS resolution failed for {host}: {e}"))?
        .collect();

    if addrs.is_empty() {
        return Err(format!("DNS returned no records for {host}"));
    }

    // ALL resolved addresses must be public.
    for addr in &addrs {
        if is_private_ip(addr.ip()) {
            return Err(format!("blocked: {host} resolves to private IP {}", addr.ip()));
        }
    }

    // Pin the URL to the first resolved IP to prevent DNS rebinding.
    let pinned_ip = addrs[0].ip();
    let mut pinned = parsed.clone();
    let ip_host = match pinned_ip {
        std::net::IpAddr::V4(v4) => v4.to_string(),
        std::net::IpAddr::V6(v6) => format!("[{v6}]"),
    };
    let _ = pinned.set_host(Some(&ip_host));
    Ok((pinned.to_string(), original_host))
}

/// Check if an IP address is private, loopback, or link-local.
fn is_private_ip(ip: std::net::IpAddr) -> bool {
    match ip {
        std::net::IpAddr::V4(v4) => {
            v4.is_loopback()             // 127.0.0.0/8
                || v4.is_private()       // 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16
                || v4.is_link_local()    // 169.254.0.0/16
                || v4.is_unspecified()   // 0.0.0.0
                || v4.is_broadcast() // 255.255.255.255
        }
        std::net::IpAddr::V6(v6) => {
            v6.is_loopback()             // ::1
                || v6.is_unspecified()   // ::
                || is_ipv6_link_local(v6) // fe80::/10
                || is_ipv6_unique_local(v6) // fc00::/7
                // Check if it's an IPv4-mapped IPv6 address (::ffff:x.x.x.x)
                || v6.to_ipv4_mapped().is_some_and(|v4| {
                    v4.is_loopback()
                        || v4.is_private()
                        || v4.is_link_local()
                        || v4.is_unspecified()
                })
        }
    }
}

/// Check for IPv6 link-local (fe80::/10).
fn is_ipv6_link_local(ip: std::net::Ipv6Addr) -> bool {
    let octets = ip.octets();
    octets[0] == 0xfe && (octets[1] & 0xc0) == 0x80
}

/// Check for IPv6 unique local (fc00::/7).
fn is_ipv6_unique_local(ip: std::net::Ipv6Addr) -> bool {
    let octets = ip.octets();
    (octets[0] & 0xfe) == 0xfc
}

/// Token bucket rate limiter.
struct TokenBucket {
    tokens: u32,
    max_tokens: u32,
    refill_interval: std::time::Duration,
    last_refill: Instant,
}

impl TokenBucket {
    fn new(max_tokens: u32, refill_interval: std::time::Duration) -> Self {
        Self {
            tokens: max_tokens,
            max_tokens,
            refill_interval,
            last_refill: Instant::now(),
        }
    }

    fn try_acquire(&mut self) -> bool {
        self.refill();
        if self.tokens > 0 {
            self.tokens -= 1;
            true
        } else {
            false
        }
    }

    fn refill(&mut self) {
        let elapsed = self.last_refill.elapsed();
        if elapsed >= self.refill_interval {
            self.tokens = self.max_tokens;
            self.last_refill = Instant::now();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn glob_match_exact() {
        assert!(glob_match(
            "https://api.example.com/v1",
            "https://api.example.com/v1"
        ));
        assert!(!glob_match(
            "https://api.example.com/v1",
            "https://api.example.com/v2"
        ));
    }

    #[test]
    fn glob_match_wildcard_suffix() {
        assert!(glob_match(
            "https://api.example.com/*",
            "https://api.example.com/v1/users"
        ));
        assert!(!glob_match(
            "https://api.example.com/*",
            "https://other.com/v1"
        ));
    }

    #[test]
    fn glob_match_wildcard_middle() {
        assert!(glob_match(
            "https://*.example.com/api",
            "https://sub.example.com/api"
        ));
    }

    #[test]
    fn sanitize_key_valid() {
        assert_eq!(PluginSandbox::sanitize_key("my-key"), Some("my-key".into()));
        assert_eq!(
            PluginSandbox::sanitize_key("data.json"),
            Some("data.json".into())
        );
    }

    #[test]
    fn sanitize_key_rejects_traversal() {
        assert!(PluginSandbox::sanitize_key("../etc/passwd").is_none());
        assert!(PluginSandbox::sanitize_key("foo/bar").is_none());
        assert!(PluginSandbox::sanitize_key("foo\\bar").is_none());
        assert!(PluginSandbox::sanitize_key("foo\0bar").is_none());
    }

    #[test]
    fn token_bucket_basic() {
        let mut bucket = TokenBucket::new(3, std::time::Duration::from_secs(60));
        assert!(bucket.try_acquire());
        assert!(bucket.try_acquire());
        assert!(bucket.try_acquire());
        assert!(!bucket.try_acquire()); // Exhausted
    }

    // ── is_private_url tests ──────────────────────────────────────

    #[test]
    fn private_url_loopback_ipv4() {
        assert!(is_private_url("http://127.0.0.1/foo"));
        assert!(is_private_url("http://127.0.0.255:8080/bar"));
        assert!(is_private_url("http://127.1.2.3/baz"));
    }

    #[test]
    fn private_url_loopback_ipv6() {
        assert!(is_private_url("http://[::1]/foo"));
        assert!(is_private_url("http://[::1]:8080/bar"));
    }

    #[test]
    fn private_url_private_ranges() {
        // 10.0.0.0/8
        assert!(is_private_url("http://10.0.0.1/"));
        assert!(is_private_url("http://10.255.255.255/"));
        // 172.16.0.0/12
        assert!(is_private_url("http://172.16.0.1/"));
        assert!(is_private_url("http://172.31.255.255/"));
        // 192.168.0.0/16
        assert!(is_private_url("http://192.168.1.1/"));
        assert!(is_private_url("http://192.168.0.0/"));
    }

    #[test]
    fn private_url_link_local() {
        // 169.254.0.0/16 (IPv4 link-local)
        assert!(is_private_url("http://169.254.1.1/"));
        assert!(is_private_url("http://169.254.169.254/")); // AWS metadata
                                                            // fe80::/10 (IPv6 link-local)
        assert!(is_private_url("http://[fe80::1]/"));
    }

    #[test]
    fn private_url_localhost_hostname() {
        assert!(is_private_url("http://localhost/foo"));
        assert!(is_private_url("http://localhost:3000/bar"));
        assert!(is_private_url("http://LOCALHOST/foo"));
        assert!(is_private_url("http://sub.localhost/foo"));
    }

    #[test]
    fn private_url_unspecified() {
        assert!(is_private_url("http://0.0.0.0/"));
        assert!(is_private_url("http://[::]/"));
    }

    #[test]
    fn private_url_ipv6_unique_local() {
        // fc00::/7
        assert!(is_private_url("http://[fc00::1]/"));
        assert!(is_private_url("http://[fd12:3456:789a::1]/"));
    }

    #[test]
    fn private_url_ipv4_mapped_ipv6() {
        // ::ffff:127.0.0.1
        assert!(is_private_url("http://[::ffff:127.0.0.1]/"));
        assert!(is_private_url("http://[::ffff:10.0.0.1]/"));
        assert!(is_private_url("http://[::ffff:192.168.1.1]/"));
    }

    #[test]
    fn private_url_unparseable() {
        assert!(is_private_url("not a url"));
        assert!(is_private_url(""));
    }

    #[test]
    fn private_url_public_ip_addresses_allowed() {
        // Use IP literals to avoid DNS dependency in tests.
        assert!(!is_private_url("https://1.1.1.1/dns-query"));
        assert!(!is_private_url("https://8.8.8.8/"));
        assert!(!is_private_url("https://93.184.216.34/")); // example.com
    }

    // ── URL allowlist integration tests ────────────────────────────

    fn make_sandbox(allowed_urls: Vec<String>, max_fetch: u32) -> PluginSandbox {
        let perms = WasmPluginPermissions {
            allowed_urls,
            max_fetch_per_minute: max_fetch,
            storage_quota_kb: 1024,
            ..Default::default()
        };
        PluginSandbox::new(
            "test-plugin".into(),
            perms,
            std::path::Path::new("/tmp/bext-test"),
        )
    }

    #[test]
    fn url_allowlist_blocks_when_empty() {
        let sandbox = make_sandbox(vec![], 10);
        assert!(!sandbox.is_url_allowed("https://api.example.com/v1"));
    }

    #[test]
    fn url_allowlist_allows_matching() {
        let sandbox = make_sandbox(vec!["https://api.example.com/*".into()], 10);
        assert!(sandbox.is_url_allowed("https://api.example.com/v1/users"));
        assert!(sandbox.is_url_allowed("https://api.example.com/v2/data"));
        assert!(!sandbox.is_url_allowed("https://evil.com/"));
    }

    #[test]
    fn url_allowlist_exact_match() {
        let sandbox = make_sandbox(vec!["https://api.example.com/v1".into()], 10);
        assert!(sandbox.is_url_allowed("https://api.example.com/v1"));
        assert!(!sandbox.is_url_allowed("https://api.example.com/v2"));
    }

    // ── Rate limiting integration tests ────────────────────────────

    #[test]
    fn rate_limit_exhaustion() {
        let sandbox = make_sandbox(vec!["https://*".into()], 3);
        assert!(sandbox.try_fetch());
        assert!(sandbox.try_fetch());
        assert!(sandbox.try_fetch());
        assert!(!sandbox.try_fetch()); // Exhausted
        assert!(!sandbox.try_fetch()); // Still exhausted
    }

    // ── KV namespace permission tests ─────────────────────────────

    #[test]
    fn kv_namespace_allowed() {
        let perms = WasmPluginPermissions {
            kv_namespaces: vec!["cache".into(), "session".into()],
            ..Default::default()
        };
        let sandbox =
            PluginSandbox::new("test".into(), perms, std::path::Path::new("/tmp/bext-test"));
        assert!(sandbox.is_kv_namespace_allowed("cache"));
        assert!(sandbox.is_kv_namespace_allowed("session"));
        assert!(!sandbox.is_kv_namespace_allowed("secret"));
        assert!(!sandbox.is_kv_namespace_allowed(""));
    }

    #[test]
    fn kv_namespace_empty_blocks_all() {
        let perms = WasmPluginPermissions::default();
        let sandbox =
            PluginSandbox::new("test".into(), perms, std::path::Path::new("/tmp/bext-test"));
        assert!(!sandbox.is_kv_namespace_allowed("anything"));
    }

    // ── Queue name permission tests ───────────────────────────────

    #[test]
    fn queue_allowed() {
        let perms = WasmPluginPermissions {
            queue_names: vec!["tasks".into(), "emails".into()],
            ..Default::default()
        };
        let sandbox =
            PluginSandbox::new("test".into(), perms, std::path::Path::new("/tmp/bext-test"));
        assert!(sandbox.is_queue_allowed("tasks"));
        assert!(sandbox.is_queue_allowed("emails"));
        assert!(!sandbox.is_queue_allowed("admin"));
        assert!(!sandbox.is_queue_allowed(""));
    }

    #[test]
    fn queue_empty_blocks_all() {
        let perms = WasmPluginPermissions::default();
        let sandbox =
            PluginSandbox::new("test".into(), perms, std::path::Path::new("/tmp/bext-test"));
        assert!(!sandbox.is_queue_allowed("anything"));
    }
}
