<file_sep>use std::{
    convert::TryInto,
    error::Error,
    fmt::{self, Display, Formatter},
};

#[derive(Debug, Clone)]
pub struct BufferTooSmall;

impl Display for BufferTooSmall {
    fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
        write!(fmt, "buffer too small")
    }
}

impl Error for BufferTooSmall {}

#[derive(Debug, Copy, Clone, PartialEq)]
pub struct Header<'a>(&'a [u8]);

impl<'a> Header<'a> {
    pub fn from_slice(buf: &'a [u8]) -> Result<Self, BufferTooSmall> {
        if buf.len() < 12 {
            return Err(BufferTooSmall);
        }

        Ok(Header(&buf[..12]))
    }

    #[inline]
    pub fn version(&self) -> u8 {
        let buf = self.as_slice();
        let byte = buf[0];
        byte >> 6
    }

    #[inline]
    pub fn ssrc(&self) -> u32 {
        u32::from_be_bytes(self.as_slice()[8..12].try_into().unwrap())
    }

    #[inline]
    pub fn as_slice(&self) -> &[u8] {
        match self {
            Header(buf) => buf,
        }
    }
}

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

    #[test]
    fn test_parse_rtp() {
        let header = Header(&[128, 96, 0, 17, 0, 0, 140, 160, 0, 0, 0, 16]);

        assert_eq!(2, header.version());
        assert_eq!(17, header.sequence_number());
        assert_eq!(36000, header.timestamp());
        assert_eq!(16, header.ssrc());
    }
}
<file_sep>#[macro_use]
extern crate clap;
#[macro_use]
extern crate log;

use std::{
    error::Error,
    io::{BufWriter, Write},
    net::{SocketAddr, TcpStream, UdpSocket},
    sync::{
        mpsc::{self, Receiver, SyncSender},
        Arc,
    },
    thread,
    time::Duration,
};

use clap::{App, AppSettings, Arg, SubCommand};
use rmpv::ValueRef;

#[derive(Debug)]
enum Address {
    Udp(SocketAddr),
    Https(String, u16),
}

impl Address {
    pub fn from_str(addr: &str) -> Result<Self, Box<dyn Error>> {
        if !addr.contains("://") {
            return Err("invalid address - must be an URL".into());
        }

        let mut it = addr.splitn(2, "://");

        let protocol = match it.next() {
            Some(protocol) => protocol,
            None => return Err("missing protocol".into()),
        };

        let addr = match it.next() {
            Some(addr) => addr,
            None => return Err("missing address".into()),
        };

        match protocol {
            "udp" => {
                let addr = addr.parse()?;
                Ok(Address::Udp(addr))
            }
            "https" => {
                let (host, port) = split_host_port(addr)?;
                Ok(Address::Https(host.into(), port))
            }
            protocol => Err(format!("unknown protocol: {}", protocol).into()),
        }
    }
}

fn split_host_port(addr: &str) -> Result<(&str, u16), Box<dyn Error>> {
    let mut it = addr.rsplitn(2, ':');
    let port = match it.next() {
        Some(port) => port.parse()?,
        None => 443,
    };

    let host = match it.next() {
        Some(host) => host,
        None => return Err("missing hostname".into()),
    };

    Ok((host, port))
}

fn main() -> Result<(), Box<dyn Error>> {
    env_logger::init();

    let matches = App::new(crate_name!())
        .version(crate_version!())
        .author(crate_authors!())
        .setting(AppSettings::SubcommandRequired)
        .subcommand(SubCommand::with_name("scan").about("scan local network for cleverdog camera(s)"))
        .subcommand(
            SubCommand::with_name("stream")
                .about("stream H264 from camera")
                .arg(
                    Arg::with_name("addr")
                        .long("addr")
                        .value_name("ADDRESS")
                        .help("network address, udp:// or https://")
                        .required(true)
                        .takes_value(true),
                )
                .arg(
                    Arg::with_name("retries")
                        .long("retries")
                        .value_name("NUMBER")
                        .default_value("18446744073709551615")
                        .help("number of retries in case of camera hanging")
                        .takes_value(true),
                ),
        )
        .get_matches();

    match matches.subcommand() {
        ("scan", ..) => {
            let info = cleverdog::lookup()?;
            println!("Address: {}", info.addr());
            println!("CID:     {}", core::str::from_utf8(info.cid())?);
            println!("MAC:     {}", info.mac());
            println!("Version: {}", info.version());
        }
        ("stream", Some(matches)) => {
            // This cannot panic because of CLAP required flag.
            let dst = matches.value_of("addr").unwrap();
            let mut num: u64 = matches.value_of("retries").unwrap().parse()?;

            let addr = Address::from_str(dst)?;
            info!("Destination address: {:?}", addr);

            let info = cleverdog::lookup()?;
            info!("Successfully resolved camera");
            info!("  Address: {}", info.addr());
            info!("  CID:     {}", core::str::from_utf8(info.cid())?);
            info!("  MAC:     {}", info.mac());
            info!("  Version: {}", info.version());

            match addr {
                Address::Udp(addr) => {
                    let sock = UdpSocket::bind("0.0.0.0:0")?;

                    cleverdog::stream(info.cid(), info.addr(), |buf| {
                        debug!("-> {}", buf.len());
                        sock.send_to(buf, addr)?;
                        Ok(())
                    })?;
                }
                Address::Https(host, port) => {
                    let addr = format!("{}:{}", host, port);

                    let (tx, rx): (SyncSender<Vec<u8>>, Receiver<Vec<u8>>) = mpsc::sync_channel(4096);

                    let thread = thread::spawn(move || {
                        let mut cfg = rustls::ClientConfig::new();
                        cfg.root_store.add_server_trust_anchors(&webpki_roots::TLS_SERVER_ROOTS);
                        let cfg = Arc::new(cfg);
                        let hostname = webpki::DNSNameRef::try_from_ascii_str(&host).expect("ASCII hostname");

                        loop {
                            let mut session = rustls::ClientSession::new(&cfg, hostname);

                            debug!("connecting to {}", addr);
                            let mut stream = match TcpStream::connect(&addr) {
                                Ok(stream) => stream,
                                Err(err) => {
                                    error!("failed to connect to {}: {}", addr, err);
                                    thread::sleep(Duration::new(1, 0));
                                    continue;
                                }
                            };

                            let mut stream = BufWriter::new(rustls::Stream::new(&mut session, &mut stream));

                            info!("successfully connected to {}", addr);

                            while let Ok(buf) = rx.recv() {
                                if let Err(err) = stream.write_all(&buf) {
                                    error!("failed to send bytes: {}", err);
                                    break;
                                }
                            }

                            thread::sleep(Duration::new(1, 0));
                        }
                    });

                    let on_data = |buf: &[u8]| {
                        debug!("-> {}", buf.len());

                        let mut msg = Vec::new();
                        if let Err(err) = rmpv::encode::write_value_ref(&mut msg, &ValueRef::Binary(buf)) {
                            error!("failed to encode datagram: {}", err);
                        }

                        if let Err(..) = tx.try_send(msg) {
                            error!("failed to send datagram due to backpressuring");
                        }

                        Ok(())
                    };

                    while num > 0 {
                        if let Err(err) = cleverdog::stream(info.cid(), info.addr(), on_data) {
                            warn!("streaming stopped: {}", err);
                        }

                        num -= 1;
                        thread::sleep(Duration::new(1, 0));
                    }

                    thread.join().unwrap();
                }
            }
        }
        (..) => unreachable!(),
    }

    Ok(())
}

// ffmpeg -protocol_whitelist file,udp,rtp -i /mnt/hls/camera.sdp -preset
// ultrafast -vcodec libx264 -r 15 -b 300k -f flv rtmp://localhost/show/camera0
<file_sep>use core::{
    fmt::{self, Display, Formatter, LowerHex, UpperHex},
    num::ParseIntError,
    str::FromStr,
};
use std::error::Error;

/// An error that can occur during parsing a MAC address string.
#[derive(Debug, Clone)]
pub enum ParseError {
    /// Parsing of the MAC address contained an invalid digit.
    InvalidDigit(ParseIntError),
    /// The MAC address did not have the correct length.
    InvalidLength,
}

impl Display for ParseError {
    fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
        match self {
            ParseError::InvalidDigit(..) => write!(fmt, "invalid digit"),
            ParseError::InvalidLength => fmt.write_str("invalid length"),
        }
    }
}

impl Error for ParseError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            ParseError::InvalidDigit(err) => Some(err),
            ParseError::InvalidLength => None,
        }
    }
}

/// MAC address.
#[derive(Debug, Clone, Copy)]
pub struct MacAddr([u8; 6]);

impl MacAddr {
    /// Constructs a new MAC address from the given fixed array.
    #[inline]
    pub fn new(buf: [u8; 6]) -> Self {
        Self(buf)
    }

    /// Parses the specified string into MAC address.
    ///
    /// ```
    /// use cleverdog::mac::MacAddr;
    ///
    /// let mac: MacAddr = "dc:a9:04:97:9d:9b".parse().unwrap();
    /// assert_eq!(mac.as_bytes(), [220, 169, 4, 151, 157, 155]);
    /// ```
    #[inline]
    pub fn from_str(s: &str) -> Result<Self, ParseError> {
        <MacAddr as FromStr>::from_str(s)
    }

    /// Returns the underlying byte array.
    #[inline]
    pub fn as_bytes(&self) -> [u8; 6] {
        match self {
            MacAddr(buf) => *buf,
        }
    }
}

impl FromStr for MacAddr {
    type Err = ParseError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let mut buf = [0u8; 6];
        let mut idx = 0;

        for b in s.split(':') {
            if idx == 6 {
                return Err(ParseError::InvalidLength);
            }

            buf[idx] = u8::from_str_radix(b, 16).map_err(|err| ParseError::InvalidDigit(err))?;
            idx += 1;
        }

        if idx != 6 {
            return Err(ParseError::InvalidLength);
        }

        Ok(MacAddr::new(buf))
    }
}

impl Display for MacAddr {
    fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
        <MacAddr as LowerHex>::fmt(self, fmt)
    }
}

impl LowerHex for MacAddr {
    fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
        write!(fmt, "{:<02x}", self.as_bytes()[0])?;

        for v in self.as_bytes().iter().skip(1) {
            write!(fmt, ":{:<02x}", v)?;
        }

        Ok(())
    }
}

impl UpperHex for MacAddr {
    fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
        write!(fmt, "{:<02X}", self.as_bytes()[0])?;

        for v in self.as_bytes().iter().skip(1) {
            write!(fmt, ":{:<02X}", v)?;
        }

        Ok(())
    }
}

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

    #[test]
    fn test_parse() {
        let v = "dc:a9:04:97:9d:9b";
        let mac = v.parse::<MacAddr>().unwrap();
        assert_eq!(mac.as_bytes(), [220, 169, 4, 151, 157, 155]);
    }

    #[test]
    fn test_display() {
        let mac = MacAddr::new([220, 169, 4, 151, 157, 155]);
        assert_eq!(&format!("{}", mac), "dc:a9:04:97:9d:9b");
    }

    #[test]
    fn test_display_lower_hex() {
        let mac = MacAddr::new([220, 169, 4, 151, 157, 155]);
        assert_eq!(&format!("{:x}", mac), "dc:a9:04:97:9d:9b");
    }

    #[test]
    fn test_display_upper_hex() {
        let mac = MacAddr::new([220, 169, 4, 151, 157, 155]);
        assert_eq!(&format!("{:X}", mac), "DC:A9:04:97:9D:9B");
    }
}
<file_sep>pub use crate::protocol::{
    scan::{LookupInfo, ScanInfo},
    version::Version,
};

mod scan;
mod version;

/// Magic constant that is prepended to each camera frame.
///
/// Represents a big-endian integer representation of `[0x4d, 0x4a]` array.
pub const MAGIC: u16 = 19786;

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

    #[test]
    fn test_magic_endianess() {
        assert_eq!([0x4d, 0x4a], MAGIC.to_be_bytes());
    }
}
<file_sep>edition = "2018"
merge_imports = true
wrap_comments = true
reorder_imports = true
max_width = 120
error_on_line_overflow = true
use_field_init_shorthand = true
<file_sep>use core::{num::ParseIntError};
use core::fmt::{self, Display, Formatter};

#[derive(Debug, Clone, Copy)]
pub struct Version([u16; 4]);

impl Version {
    #[inline]
    pub fn new(buf: [u16; 4]) -> Self {
        Self(buf)
    }

    pub fn from_str(v: &str) -> Result<Self, ParseIntError> {
        let mut buf = [0u16; 4];
        let mut idx = 0;

        for b in v.split('.') {
            if idx == 4 {
                break;
            }

            buf[idx] = u16::from_str_radix(b, 10)?;
            idx += 1;
        }

        Ok(Self::new(buf))
    }
}

impl Display for Version {
    fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
        let Version([major, minor, patch, release]) = self;
        write!(fmt, "{}.{}.{}.{}", major, minor, patch, release)
    }
}
<file_sep>[package]
name = "cleverdog"
version = "0.1.1"
authors = ["<NAME> <<EMAIL>>"]
edition = "2018"
description = "Cleverdog Camera API and basic streaming proxy"
license = "MIT"

[dependencies]
byteorder = "1"
log = "0.4"

[dev-dependencies]
clap = "2"
env_logger = "0.6"
rustls = "0.15"
webpki = "0.19"
webpki-roots = "0.16"
rmpv = "0.4"


[profile.release]
panic = "abort"
<file_sep>use core::{convert::TryFrom, str};
use std::net::SocketAddr;

use crate::{mac::MacAddr, protocol::version::Version};

#[derive(Debug, Clone, Copy)]
pub struct ScanInfo {
    /// Camera MAC address.
    mac: MacAddr,
    /// Firmware version.
    version: Version,
}

impl TryFrom<&[u8]> for ScanInfo {
    type Error = &'static str;

    fn try_from(v: &[u8]) -> Result<Self, Self::Error> {
        let mut it = v.split(|&ch| ch == b'\0');

        let mac = match it.next() {
            Some(mac) => match str::from_utf8(mac.into()) {
                Ok(mac) => match MacAddr::from_str(mac) {
                    Ok(mac) => mac,
                    Err(..) => return Err("MAC address is invalid"),
                },
                Err(..) => return Err("MAC address contains invalid UTF-8 sequence"),
            },
            None => return Err("missing MAC address"),
        };

        let version = match it.next() {
            Some(version) => match str::from_utf8(version.into()) {
                Ok(version) => match Version::from_str(version) {
                    Ok(version) => version,
                    Err(..) => return Err("version is invalid"),
                },
                Err(..) => return Err("version contains invalid UTF-8 sequence"),
            },
            None => return Err("missing version"),
        };

        let v = Self { mac, version };

        Ok(v)
    }
}

#[derive(Debug, Clone, Copy)]
pub struct LookupInfo {
    /// Camera endpoint.
    addr: SocketAddr,
    /// Camera ID.
    cid: [u8; 16],
    /// Scan info.
    info: ScanInfo,
}

impl LookupInfo {
    pub fn new(addr: SocketAddr, cid: [u8; 16], info: ScanInfo) -> Self {
        Self { addr, cid, info }
    }

    /// Returns socket address where the camera is bound.
    #[inline]
    pub fn addr(&self) -> SocketAddr {
        self.addr
    }

    /// Returns camera's client id.
    #[inline]
    pub fn cid(&self) -> &[u8] {
        &self.cid[..]
    }

    /// Returns camera's MAC address.
    #[inline]
    pub fn mac(&self) -> &MacAddr {
        &self.info.mac
    }

    /// Returns camera's firmware version.
    #[inline]
    pub fn version(&self) -> &Version {
        &self.info.version
    }
}
<file_sep>use core::{convert::TryFrom, time::Duration};
use std::{
    error::Error,
    io::{Cursor, ErrorKind, Read, Write},
    net::{SocketAddr, UdpSocket},
    time::{Instant, SystemTime},
};

use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};

use crate::{
    protocol::{LookupInfo, ScanInfo, MAGIC},
    rtp::Header,
};

pub mod mac;
pub mod protocol;
mod rtp;

enum Command {
    Scan,
    ScanReply,
    StartRtp,
}

impl Command {
    pub fn as_u16(&self) -> u16 {
        match self {
            Command::Scan => 0x1004,
            Command::ScanReply => 0x100e,
            Command::StartRtp => 0x1007,
        }
    }

    pub fn encode(&self, cid: &[u8], args: &[u8]) -> Result<Vec<u8>, Box<dyn Error>> {
        let mut cid = cid;
        if cid.len() > 15 {
            cid = &cid[..15]
        }

        let mut buf = Cursor::new(Vec::new());

        buf.write_u16::<BigEndian>(MAGIC)?;
        buf.write_u16::<BigEndian>(self.as_u16())?;
        buf.write_all(cid)?;
        buf.write_all(&b"000000000000000"[..15usize.saturating_sub(cid.len())])?;
        buf.write_all(&[0x0])?;
        buf.write_all(args)?;

        Ok(buf.into_inner())
    }
}

pub fn lookup() -> Result<LookupInfo, Box<dyn Error>> {
    let sock = UdpSocket::bind("0.0.0.0:0")?;
    sock.set_broadcast(true)?;
    sock.set_read_timeout(Some(Duration::new(1, 0)))?;

    let comm = Command::Scan.encode(b"", b"00000000000000000000000000000000000000")?;
    sock.send_to(&comm, "192.168.1.71:10008")?;

    let mut buf = [0; 4096];

    loop {
        let (size, addr) = match sock.recv_from(&mut buf[..]) {
            Ok((size, addr)) => (size, addr),
            Err(ref err) if err.kind() == ErrorKind::WouldBlock => {
                return Err("timed out".into());
            }
            Err(err) => return Err(err.into()),
        };

        let mut buf = Cursor::new(&buf[..size]);

        let magic = buf.read_u16::<BigEndian>()?;

        if magic != MAGIC {
            return Err("invalid magic header".into());
        }

        let comm = buf.read_u16::<BigEndian>()?;

        if comm != Command::ScanReply.as_u16() {
            continue;
        }

        let mut cid = [0; 16];
        buf.read_exact(&mut cid[..])?;

        let idx = buf.position() as usize;
        let info = ScanInfo::try_from(&buf.into_inner()[idx..])?;
        let info = LookupInfo::new(addr, cid, info);

        return Ok(info);
    }
}

pub fn stream<F>(cid: &[u8], src: SocketAddr, f: F) -> Result<(), Box<dyn Error>>
where
    F: Fn(&[u8]) -> Result<(), Box<dyn Error>>,
{
    let sock = UdpSocket::bind("0.0.0.0:0")?;
    sock.set_read_timeout(Some(Duration::new(10, 0)))?;

    let local_addr = sock.local_addr()?;

    let mut args = Cursor::new(Vec::new());
    args.write_all(b"00000000000000000000000000000000000000")?;
    args.write_fmt(format_args!("{}:{}\0", local_addr.port(), local_addr.port()))?;

    let comm = Command::StartRtp.encode(cid, &args.into_inner())?;
    sock.send_to(&comm, src)?;

    let mut timestamp = Instant::now();
    let mut buf = [0; 4096];
    loop {
        let (size, addr) = sock.recv_from(&mut buf[..])?;

        if timestamp.elapsed() >= Duration::from_secs(1) {
            timestamp = Instant::now();
            send_rtcp(&sock, &addr)?;
        }

        if buf[..size].len() < 16 {
            continue;
        }

        let hdr = Header::from_slice(&buf[4..])?;

        if hdr.version() != 2 {
            continue;
        }

        // Skip non-video frames.
        if buf[2] != 1 {
            continue;
        }

        if hdr.ssrc() != 16 {
            continue;
        }

        f(&buf[4..size])?;
    }
}

fn send_rtcp(sock: &UdpSocket, camera: &SocketAddr) -> Result<(), Box<dyn Error>> {
    let mut buf = Cursor::new(Vec::new());

    buf.write_all(&[
        0x00, 0x00, 0x01, 0x00, // Header.
        0x80, // RTP v2
        0xc8, // RTCP sender report packet type
        0x00, 0x06,
    ])?;
    buf.write_u32::<BigEndian>(0x00000002)?;

    let msecs = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?.as_nanos() / 1e6 as u128 + 2208988800000;
    let seconds = (msecs / 1000) as u32;
    let fraction = (0x100000000 * (msecs % 1000) / 1000) as u32;

    buf.write_u32::<BigEndian>(seconds)?;
    buf.write_u32::<BigEndian>(fraction)?;
    buf.write_u32::<BigEndian>(0)?;
    buf.write_u32::<BigEndian>(0)?;
    buf.write_u32::<BigEndian>(0)?;

    sock.send_to(&buf.into_inner(), camera)?;

    Ok(())
}
