fluidattacks-core-git 0.70.0

use std::fmt;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};

use ipnet::{Ipv4Net, Ipv6Net};

use crate::errors::HostError;

/// Why an address is not one to reach from outside.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum LocalAddress {
    #[error("Requests to loopback addresses are not allowed")]
    Loopback,
    #[error("Requests to the local network are not allowed")]
    LocalNetwork,
}

#[derive(Debug, Clone, Copy, Default)]
pub struct AllowedLocal {
    pub localhost: bool,
    pub local_network: bool,
}

impl AllowedLocal {
    const fn lets_through(self, local: LocalAddress) -> bool {
        match local {
            LocalAddress::Loopback => self.localhost,
            LocalAddress::LocalNetwork => self.local_network,
        }
    }
}

pub enum Host<'a> {
    Ip(IpAddr),
    Name(&'a str),
}

impl fmt::Display for Host<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Ip(ip) => ip.fmt(f),
            Self::Name(name) => f.write_str(name),
        }
    }
}

const LOOPBACK_V4: Ipv4Net = Ipv4Net::new_assert(Ipv4Addr::new(127, 0, 0, 0), 8);

// IANA's special-purpose ranges, as Python's ipaddress reads them; std's is_global, which covers them, is nightly-only.
const LOCAL_NETWORK_V4: [Ipv4Net; 14] = [
    Ipv4Net::new_assert(Ipv4Addr::UNSPECIFIED, 8),
    Ipv4Net::new_assert(Ipv4Addr::new(10, 0, 0, 0), 8),
    Ipv4Net::new_assert(Ipv4Addr::new(100, 64, 0, 0), 10),
    Ipv4Net::new_assert(Ipv4Addr::new(169, 254, 0, 0), 16),
    Ipv4Net::new_assert(Ipv4Addr::new(172, 16, 0, 0), 12),
    Ipv4Net::new_assert(Ipv4Addr::new(192, 0, 0, 0), 24),
    Ipv4Net::new_assert(Ipv4Addr::new(192, 0, 0, 170), 31),
    Ipv4Net::new_assert(Ipv4Addr::new(192, 0, 2, 0), 24),
    Ipv4Net::new_assert(Ipv4Addr::new(192, 168, 0, 0), 16),
    Ipv4Net::new_assert(Ipv4Addr::new(198, 18, 0, 0), 15),
    Ipv4Net::new_assert(Ipv4Addr::new(198, 51, 100, 0), 24),
    Ipv4Net::new_assert(Ipv4Addr::new(203, 0, 113, 0), 24),
    Ipv4Net::new_assert(Ipv4Addr::new(240, 0, 0, 0), 4),
    Ipv4Net::new_assert(Ipv4Addr::BROADCAST, 32),
];

const GLOBAL_V4: [Ipv4Addr; 2] = [Ipv4Addr::new(192, 0, 0, 9), Ipv4Addr::new(192, 0, 0, 10)];

const LOCAL_NETWORK_V6: [Ipv6Net; 9] = [
    Ipv6Net::new_assert(Ipv6Addr::UNSPECIFIED, 128),
    Ipv6Net::new_assert(Ipv6Addr::new(0x64, 0xff9b, 1, 0, 0, 0, 0, 0), 48),
    Ipv6Net::new_assert(Ipv6Addr::new(0x100, 0, 0, 0, 0, 0, 0, 0), 64),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0, 0, 0, 0, 0, 0, 0), 23),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 0), 32),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2002, 0, 0, 0, 0, 0, 0, 0), 16),
    Ipv6Net::new_assert(Ipv6Addr::new(0x3fff, 0, 0, 0, 0, 0, 0, 0), 20),
    Ipv6Net::new_assert(Ipv6Addr::new(0xfc00, 0, 0, 0, 0, 0, 0, 0), 7),
    Ipv6Net::new_assert(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 0), 10),
];

const GLOBAL_V6: [Ipv6Net; 6] = [
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 1, 0, 0, 0, 0, 0, 1), 128),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 1, 0, 0, 0, 0, 0, 2), 128),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 3, 0, 0, 0, 0, 0, 0), 32),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 4, 0x112, 0, 0, 0, 0, 0), 48),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0x20, 0, 0, 0, 0, 0, 0), 28),
    Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0x30, 0, 0, 0, 0, 0, 0), 28),
];

/// `None` for an address reachable from outside; an IPv4 mapped into IPv6 counts as its IPv4.
#[must_use]
pub fn local_address(ip: IpAddr) -> Option<LocalAddress> {
    match ip {
        IpAddr::V4(ip) => local_v4(ip),
        IpAddr::V6(ip) => ip.to_ipv4_mapped().map_or_else(|| local_v6(ip), local_v4),
    }
}

pub async fn outside_addresses(
    host: &Host<'_>,
    allowed: AllowedLocal,
) -> Result<Vec<IpAddr>, HostError> {
    let ips = match host {
        Host::Ip(ip) => vec![*ip],
        Host::Name(name) => tokio::net::lookup_host((*name, 0))
            .await
            .map_err(HostError::Unresolved)?
            .map(|address| address.ip())
            .collect(),
    };
    if ips.is_empty() {
        return Err(HostError::NoAddresses);
    }
    let refused = ips
        .iter()
        .filter_map(|ip| local_address(*ip))
        .find(|local| !allowed.lets_through(*local));
    refused.map_or(Ok(ips), |local| Err(HostError::Local(local)))
}

fn local_v4(ip: Ipv4Addr) -> Option<LocalAddress> {
    if LOOPBACK_V4.contains(&ip) {
        Some(LocalAddress::Loopback)
    } else if LOCAL_NETWORK_V4.iter().any(|net| net.contains(&ip)) && !GLOBAL_V4.contains(&ip) {
        Some(LocalAddress::LocalNetwork)
    } else {
        None
    }
}

fn local_v6(ip: Ipv6Addr) -> Option<LocalAddress> {
    if ip.is_loopback() {
        Some(LocalAddress::Loopback)
    } else if LOCAL_NETWORK_V6.iter().any(|net| net.contains(&ip))
        && !GLOBAL_V6.iter().any(|net| net.contains(&ip))
    {
        Some(LocalAddress::LocalNetwork)
    } else {
        None
    }
}

#[cfg(test)]
mod tests {
    use std::net::IpAddr;

    use super::{local_address, LocalAddress};

    fn kind(ip: &str) -> Option<LocalAddress> {
        local_address(ip.parse::<IpAddr>().unwrap())
    }

    // Each address read by Python 3.13's ipaddress: is_loopback, or is_private,
    // is_link_local, 100.64.0.0/10 or 255.255.255.255.
    #[test]
    fn a_loopback_address_is_loopback() {
        for ip in ["127.0.0.1", "127.255.0.1", "::1", "::ffff:127.0.0.1"] {
            assert_eq!(kind(ip), Some(LocalAddress::Loopback), "{ip}");
        }
    }

    #[test]
    fn a_special_purpose_address_is_local_network() {
        for ip in [
            "0.0.0.0",
            "10.0.0.1",
            "100.64.0.1",
            "169.254.1.1",
            "172.16.0.1",
            "192.0.0.1",
            "192.0.0.171",
            "192.0.2.1",
            "192.168.0.1",
            "198.19.0.1",
            "198.51.100.1",
            "203.0.113.1",
            "240.0.0.1",
            "255.255.255.255",
            "::",
            "::ffff:10.0.0.1",
            "64:ff9b:1::1",
            "100::1",
            "2001:10::1",
            "2001:db8::1",
            "2002::1",
            "3fff::1",
            "fc00::1",
            "fe80::1",
        ] {
            assert_eq!(kind(ip), Some(LocalAddress::LocalNetwork), "{ip}");
        }
    }

    // Python reads 100.64.0.0/10 on the plain IPv4 only, and lets the mapped one through.
    #[test]
    fn a_mapped_ipv4_is_read_as_its_ipv4() {
        assert_eq!(kind("::ffff:100.64.0.1"), Some(LocalAddress::LocalNetwork));
    }

    #[test]
    fn a_global_address_is_not_local() {
        for ip in [
            "8.8.8.8",
            "192.0.0.9",
            "192.0.0.10",
            "224.0.0.1",
            "::ffff:8.8.8.8",
            "::127.0.0.1",
            "64:ff9b::a00:1",
            "2001:1::1",
            "2001:3::1",
            "2001:4:112::1",
            "2001:20::1",
            "2001:30::1",
            "2606:4700::1111",
        ] {
            assert_eq!(kind(ip), None, "{ip}");
        }
    }
}