arcly-stream 0.5.0

//! ONVIF device discovery via **WS-Discovery** (feature `onvif`).
//!
//! ONVIF cameras announce themselves with WS-Discovery — SOAP-over-UDP on the
//! multicast group `239.255.255.250:3702`. [`OnvifDiscovery::probe`] multicasts a
//! `Probe` for `NetworkVideoTransmitter` devices, collects the `ProbeMatch`
//! replies for a short window, and returns each device's endpoint reference,
//! device-service URLs (`XAddrs`), scopes (name/hardware/location), and types.
//!
//! From a discovered [`OnvifDevice::xaddrs`] a host can call the ONVIF device
//! service (`GetStreamUri`) to obtain the camera's `rtsp://` URL and hand it to
//! [`RtspSource`](crate::protocol::rtsp::RtspSource) — that SOAP call is the
//! host's concern; this module covers discovery only.
//!
//! The SOAP/XML is built and parsed by hand (a tiny namespace-agnostic element
//! reader), so discovery pulls in **no XML or SOAP dependency** and uses no
//! `unsafe`.
//!
//! ```no_run
//! # async fn run() -> arcly_stream::Result<()> {
//! use arcly_stream::onvif::OnvifDiscovery;
//! for cam in OnvifDiscovery::new().probe().await? {
//!     println!("{} -> {:?}", cam.best_name().as_deref().unwrap_or("camera"), cam.xaddrs);
//! }
//! # Ok(()) }
//! ```

use std::collections::HashMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::time::{Duration, Instant};

use tokio::net::UdpSocket;
use tokio::time::timeout;

use crate::Result;

/// The WS-Discovery IPv4 multicast group and port.
const WS_DISCOVERY_ADDR: (Ipv4Addr, u16) = (Ipv4Addr::new(239, 255, 255, 250), 3702);
/// Default time to collect `ProbeMatch` replies.
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(3);

/// A camera/device discovered via WS-Discovery.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OnvifDevice {
    /// The device's endpoint reference (`urn:uuid:…`), a stable identity.
    pub endpoint: String,
    /// Device-service URLs (the ONVIF `device_service` endpoints).
    pub xaddrs: Vec<String>,
    /// Advertised scopes (e.g. `onvif://www.onvif.org/name/Acme`,
    /// `…/hardware/…`, `…/location/…`).
    pub scopes: Vec<String>,
    /// Advertised device types (e.g. `dn:NetworkVideoTransmitter`).
    pub types: String,
}

impl OnvifDevice {
    /// The value of the `…/name/<value>` scope, if advertised (the camera's
    /// friendly name), with `%xx` percent-escapes decoded.
    pub fn best_name(&self) -> Option<String> {
        self.scope_value("name")
    }

    /// The value of the `…/hardware/<value>` scope, if advertised.
    pub fn hardware(&self) -> Option<String> {
        self.scope_value("hardware")
    }

    /// Decode the value of the first scope of category `category`
    /// (`onvif://www.onvif.org/<category>/<value>`).
    fn scope_value(&self, category: &str) -> Option<String> {
        let needle = format!("/{category}/");
        self.scopes.iter().find_map(|s| {
            s.find(&needle)
                .map(|i| percent_decode(&s[i + needle.len()..]))
        })
    }
}

/// A WS-Discovery prober for ONVIF cameras.
pub struct OnvifDiscovery {
    timeout: Duration,
    target: SocketAddr,
}

impl Default for OnvifDiscovery {
    fn default() -> Self {
        Self::new()
    }
}

impl OnvifDiscovery {
    /// A prober with the default 3-second collection window targeting the
    /// standard WS-Discovery multicast group.
    pub fn new() -> Self {
        Self {
            timeout: DEFAULT_TIMEOUT,
            target: WS_DISCOVERY_ADDR.into(),
        }
    }

    /// Set how long to collect `ProbeMatch` replies.
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// Override the probe destination (default: the WS-Discovery multicast
    /// group). Useful to target a specific subnet's group address or, in tests,
    /// a known responder.
    pub fn with_target(mut self, target: SocketAddr) -> Self {
        self.target = target;
        self
    }

    /// Send a `Probe` and collect the cameras that answer within the timeout,
    /// de-duplicated by endpoint reference.
    pub async fn probe(&self) -> Result<Vec<OnvifDevice>> {
        let socket = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await?;
        // Replies are unicast back to our source port; a small multicast TTL lets
        // the probe cross a hop or two of the local network.
        let _ = socket.set_multicast_ttl_v4(2);

        let message_id = format!("urn:uuid:{}", random_uuid());
        let probe = build_probe(&message_id);
        socket.send_to(probe.as_bytes(), self.target).await?;

        let mut devices: HashMap<String, OnvifDevice> = HashMap::new();
        let deadline = Instant::now() + self.timeout;
        let mut buf = vec![0u8; 16 * 1024];
        loop {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                break;
            }
            match timeout(remaining, socket.recv_from(&mut buf)).await {
                Ok(Ok((n, _from))) => {
                    let xml = String::from_utf8_lossy(&buf[..n]);
                    for dev in parse_probe_matches(&xml) {
                        devices.entry(dev.endpoint.clone()).or_insert(dev);
                    }
                }
                // Socket error or the collection window elapsed.
                _ => break,
            }
        }
        Ok(devices.into_values().collect())
    }
}

/// Build a WS-Discovery `Probe` for ONVIF `NetworkVideoTransmitter` devices.
fn build_probe(message_id: &str) -> String {
    format!(
        "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
         <e:Envelope xmlns:e=\"http://www.w3.org/2003/05/soap-envelope\" \
         xmlns:w=\"http://schemas.xmlsoap.org/ws/2004/08/addressing\" \
         xmlns:d=\"http://schemas.xmlsoap.org/ws/2005/04/discovery\" \
         xmlns:dn=\"http://www.onvif.org/ver10/network/wsdl\">\
         <e:Header>\
         <w:MessageID>{message_id}</w:MessageID>\
         <w:To e:mustUnderstand=\"true\">urn:schemas-xmlsoap-org:ws:2005:04:discovery</w:To>\
         <w:Action e:mustUnderstand=\"true\">\
         http://schemas.xmlsoap.org/ws/2005/04/discovery/Probe</w:Action>\
         </e:Header>\
         <e:Body>\
         <d:Probe>\
         <d:Types>dn:NetworkVideoTransmitter</d:Types>\
         </d:Probe>\
         </e:Body>\
         </e:Envelope>"
    )
}

/// Parse every `ProbeMatch` in a SOAP response envelope into devices. A device
/// must carry at least one `XAddrs` URL to be returned.
fn parse_probe_matches(xml: &str) -> Vec<OnvifDevice> {
    let mut out = Vec::new();
    for block in elements(xml, "ProbeMatch") {
        let xaddrs: Vec<String> = element_text(block, "XAddrs")
            .map(|s| s.split_whitespace().map(str::to_owned).collect())
            .unwrap_or_default();
        if xaddrs.is_empty() {
            continue;
        }
        // `Address` lives inside `EndpointReference`; fall back to the first
        // XAddr as the identity if a device omits it.
        let endpoint = element_text(block, "Address")
            .map(|s| s.trim().to_owned())
            .filter(|s| !s.is_empty())
            .unwrap_or_else(|| xaddrs[0].clone());
        let scopes = element_text(block, "Scopes")
            .map(|s| s.split_whitespace().map(str::to_owned).collect())
            .unwrap_or_default();
        let types = element_text(block, "Types")
            .map(|s| s.split_whitespace().collect::<Vec<_>>().join(" "))
            .unwrap_or_default();
        out.push(OnvifDevice {
            endpoint,
            xaddrs,
            scopes,
            types,
        });
    }
    out
}

// ── Minimal namespace-agnostic XML reading ──────────────────────────────────
//
// WS-Discovery responses use varied namespace prefixes (`wsa:`, `a:`, `d:`,
// `tds:`, …), so elements are matched by *local* name (the part after the last
// `:`). These readers handle the leaf/flat structure of a ProbeMatch — they are
// not a general XML parser.

/// The text content of the first element whose local name is `local`.
fn element_text<'a>(xml: &'a str, local: &str) -> Option<&'a str> {
    let (_, content) = find_element(xml, local)?;
    Some(content)
}

/// The full inner XML of every element whose local name is `local`.
fn elements<'a>(xml: &'a str, local: &str) -> Vec<&'a str> {
    let mut out = Vec::new();
    let mut rest = xml;
    while let Some((after, content)) = find_element(rest, local) {
        out.push(content);
        rest = after;
    }
    out
}

/// Find the first element with local name `local`; return `(remainder_after_it,
/// inner_content)`. Skips closing/comment/declaration tags.
fn find_element<'a>(xml: &'a str, local: &str) -> Option<(&'a str, &'a str)> {
    let mut search = 0;
    while let Some(rel) = xml[search..].find('<') {
        let open = search + rel;
        let after = &xml[open + 1..];
        // Move the cursor past this `<` so we always make progress.
        search = open + 1;
        let first = after.as_bytes().first().copied().unwrap_or(b' ');
        if first == b'/' || first == b'!' || first == b'?' {
            continue;
        }
        let name_end = after
            .find(|c: char| c == '>' || c == '/' || c.is_whitespace())
            .unwrap_or(after.len());
        if local_name(&after[..name_end]) != local {
            continue;
        }
        let gt = after.find('>')?;
        // Self-closing element: `<x .../>` — empty content.
        if after.as_bytes()[gt.saturating_sub(1)] == b'/' {
            return Some((&after[gt + 1..], ""));
        }
        let content_start = open + 1 + gt + 1;
        let tail = &xml[content_start..];
        let (content_end, after_close) = find_close(tail, local)?;
        return Some((&tail[after_close..], &tail[..content_end]));
    }
    None
}

/// Locate the matching close tag for local name `local` in `tail`; return
/// `(content_end, after_close)` byte offsets within `tail`.
fn find_close(tail: &str, local: &str) -> Option<(usize, usize)> {
    let mut search = 0;
    while let Some(rel) = tail[search..].find("</") {
        let c = search + rel;
        let after = &tail[c + 2..];
        let end = after.find('>')?;
        if local_name(after[..end].trim()) == local {
            return Some((c, c + 2 + end + 1));
        }
        search = c + 2;
    }
    None
}

/// The local part of a possibly-prefixed XML name (`wsa:Address` → `Address`).
fn local_name(name: &str) -> &str {
    name.rsplit(':').next().unwrap_or(name)
}

/// Decode `%xx` percent-escapes in an ONVIF scope value (names/locations are
/// percent-encoded). Invalid escapes are left as-is.
fn percent_decode(s: &str) -> String {
    let bytes = s.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'%' && i + 2 < bytes.len() {
            if let (Some(h), Some(l)) = (hex_val(bytes[i + 1]), hex_val(bytes[i + 2])) {
                out.push(h << 4 | l);
                i += 3;
                continue;
            }
        }
        out.push(bytes[i]);
        i += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}

fn hex_val(b: u8) -> Option<u8> {
    match b {
        b'0'..=b'9' => Some(b - b'0'),
        b'a'..=b'f' => Some(b - b'a' + 10),
        b'A'..=b'F' => Some(b - b'A' + 10),
        _ => None,
    }
}

/// A random UUID (version-4 shape) for the probe's `MessageID`. Seeded from the
/// OS-backed `RandomState`, so it's unique per probe without an RNG dependency.
fn random_uuid() -> String {
    use std::collections::hash_map::RandomState;
    use std::hash::{BuildHasher, Hasher};
    use std::time::{SystemTime, UNIX_EPOCH};

    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let mut b = [0u8; 16];
    for (i, chunk) in b.chunks_mut(8).enumerate() {
        let mut h = RandomState::new().build_hasher();
        h.write_usize(i);
        h.write_u128(now);
        chunk.copy_from_slice(&h.finish().to_le_bytes());
    }
    b[6] = (b[6] & 0x0F) | 0x40; // version 4
    b[8] = (b[8] & 0x3F) | 0x80; // variant 1
    format!(
        "{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-\
         {:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
        b[0],
        b[1],
        b[2],
        b[3],
        b[4],
        b[5],
        b[6],
        b[7],
        b[8],
        b[9],
        b[10],
        b[11],
        b[12],
        b[13],
        b[14],
        b[15]
    )
}

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

    /// A realistic ONVIF `ProbeMatch` (Hikvision-style), namespace prefixes and
    /// percent-encoded scopes included.
    const PROBE_MATCH: &str = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
        <SOAP-ENV:Envelope xmlns:SOAP-ENV=\"http://www.w3.org/2003/05/soap-envelope\" \
        xmlns:wsa=\"http://schemas.xmlsoap.org/ws/2004/08/addressing\" \
        xmlns:d=\"http://schemas.xmlsoap.org/ws/2005/04/discovery\">\
        <SOAP-ENV:Header>\
        <wsa:MessageID>urn:uuid:aabbccdd-0011-2233-4455-66778899aabb</wsa:MessageID>\
        <wsa:RelatesTo>urn:uuid:probe-1</wsa:RelatesTo>\
        </SOAP-ENV:Header>\
        <SOAP-ENV:Body>\
        <d:ProbeMatches>\
        <d:ProbeMatch>\
        <wsa:EndpointReference>\
        <wsa:Address>urn:uuid:2419d68a-2dd2-21b2-a205-1234567890ab</wsa:Address>\
        </wsa:EndpointReference>\
        <d:Types>dn:NetworkVideoTransmitter tds:Device</d:Types>\
        <d:Scopes>onvif://www.onvif.org/name/Acme%20Cam onvif://www.onvif.org/hardware/DS-2CD \
        onvif://www.onvif.org/location/Lobby</d:Scopes>\
        <d:XAddrs>http://192.168.1.64/onvif/device_service \
        http://10.0.0.5/onvif/device_service</d:XAddrs>\
        <d:MetadataVersion>1</d:MetadataVersion>\
        </d:ProbeMatch>\
        </d:ProbeMatches>\
        </SOAP-ENV:Body>\
        </SOAP-ENV:Envelope>";

    #[test]
    fn probe_targets_network_video_transmitter() {
        let p = build_probe("urn:uuid:test-id");
        assert!(p.contains("urn:uuid:test-id"));
        assert!(p.contains("dn:NetworkVideoTransmitter"));
        assert!(p.contains(".../discovery/Probe") || p.contains("discovery/Probe"));
    }

    #[test]
    fn parses_a_real_probe_match() {
        let devs = parse_probe_matches(PROBE_MATCH);
        assert_eq!(devs.len(), 1);
        let d = &devs[0];
        assert_eq!(d.endpoint, "urn:uuid:2419d68a-2dd2-21b2-a205-1234567890ab");
        assert_eq!(
            d.xaddrs,
            vec![
                "http://192.168.1.64/onvif/device_service",
                "http://10.0.0.5/onvif/device_service",
            ]
        );
        assert_eq!(d.types, "dn:NetworkVideoTransmitter tds:Device");
        assert_eq!(d.best_name().as_deref(), Some("Acme Cam")); // %20 decoded
        assert_eq!(d.hardware().as_deref(), Some("DS-2CD"));
    }

    #[test]
    fn skips_matches_without_xaddrs() {
        let no_addr = "<d:ProbeMatch><wsa:Address>urn:uuid:x</wsa:Address></d:ProbeMatch>";
        assert!(parse_probe_matches(no_addr).is_empty());
    }

    #[test]
    fn element_reader_is_namespace_agnostic_and_handles_multiples() {
        let xml = "<a:Foo>one</a:Foo><Foo>two</Foo>";
        assert_eq!(element_text(xml, "Foo"), Some("one"));
        assert_eq!(elements(xml, "Foo"), vec!["one", "two"]);
        assert_eq!(element_text(xml, "Bar"), None);
    }

    #[test]
    fn self_closing_and_unterminated_are_safe() {
        assert_eq!(element_text("<x:Empty/>", "Empty"), Some(""));
        assert_eq!(element_text("<x:Open>no close", "Open"), None);
    }

    #[test]
    fn uuid_has_v4_shape_and_varies() {
        let a = random_uuid();
        assert_eq!(a.len(), 36);
        assert_eq!(a.as_bytes()[14], b'4', "version nibble");
        assert!(matches!(a.as_bytes()[19], b'8' | b'9' | b'a' | b'b'));
        assert_ne!(a, random_uuid());
    }

    /// End-to-end over loopback: a fake "camera" answers a probe with a
    /// `ProbeMatch`, and discovery parses it into a device.
    #[tokio::test]
    async fn discovers_a_responding_device() {
        let camera = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
        let cam_addr = camera.local_addr().unwrap();

        // The camera: wait for a probe, reply with a ProbeMatch to the prober.
        tokio::spawn(async move {
            let mut buf = [0u8; 4096];
            let (n, from) = camera.recv_from(&mut buf).await.unwrap();
            assert!(String::from_utf8_lossy(&buf[..n]).contains("Probe"));
            camera.send_to(PROBE_MATCH.as_bytes(), from).await.unwrap();
        });

        let devices = OnvifDiscovery::new()
            .with_timeout(Duration::from_secs(2))
            .with_target(cam_addr)
            .probe()
            .await
            .unwrap();

        assert_eq!(devices.len(), 1);
        assert_eq!(devices[0].best_name().as_deref(), Some("Acme Cam"));
    }
}