Behavision/behavision/discover.py

"""Find the cameras on the shop's network, so nobody has to type an address.

The add-camera form asked for an IP address, and a shop owner does not know
their camera's IP address. It is on a sticker under the camera, if at all, or
inside the camera's own app under a menu called something different for every
make. That one field is where onboarding stopped for anyone who was not an
installer.

Two probes, merged:

- **WS-Discovery** (ONVIF's discovery protocol): one multicast to
  239.255.255.250:3702 and every ONVIF camera on the LAN answers with its
  address and, usually, its make and model. Cheap, fast, and names the device
  - but only cameras that speak ONVIF answer, and some cheap ones do not.
- **A TCP sweep of port 554** across the local /24: anything listening on the
  RTSP port is very probably a camera or a recorder. Names nothing, misses
  nothing that streams.

A host found by either is a candidate; one found by both is a camera with a
name. The result is a list to pick from, not a decision: the person still
supplies the password, and Test still proves the stream opens.

Stdlib only. This runs inside the engine, which ships as a small wheel, and a
network-scanning dependency would be a large thing to add for two sockets.
"""
from __future__ import annotations

import ipaddress
import re
import socket
import uuid
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field, asdict
from typing import Iterable

# ONVIF WS-Discovery probe. The MessageID must be unique per probe; devices
# ignore a repeat.
_PROBE = (
    '<?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>uuid:{mid}</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>'
)
_MCAST = ("239.255.255.250", 3702)

# Makes we can name from an ONVIF scope or hostname, mapped to the ids the
# camera-make picker uses so the form can preselect the stream path.
_MAKES = (
    ("hikvision", "hikvision"), ("hik", "hikvision"), ("dahua", "dahua"),
    ("cp plus", "cpplus"), ("cpplus", "cpplus"), ("cp-plus", "cpplus"),
    ("uniview", "uniview"), ("unv", "uniview"), ("tapo", "tplink"),
    ("tp-link", "tplink"), ("reolink", "reolink"), ("amcrest", "amcrest"),
    ("axis", "axis"),
)


@dataclass
class Found:
    host: str
    rtsp: bool = False            # port 554 answered
    onvif: bool = False           # answered WS-Discovery
    name: str = ""                # from ONVIF scopes, e.g. "Hikvision DS-2CD2043"
    make: str = ""                # picker id, when it can be guessed
    onvif_url: str = ""
    sources: list[str] = field(default_factory=list)


def local_networks() -> list[ipaddress.IPv4Network]:
    """The /24s this machine sits on, best effort and without dependencies.

    Interface masks are not portable in the stdlib, so this assumes /24 - the
    shape of nearly every shop's router - for each local IPv4 address it can
    find. A bigger network would need a scan anyway that this should not run
    unasked.
    """
    addrs: set[str] = set()
    try:
        # The address the OS would use to reach the internet: the LAN we care about.
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.settimeout(0.5)
        s.connect(("8.8.8.8", 80))
        addrs.add(s.getsockname()[0])
        s.close()
    except OSError:
        pass
    try:
        for a in socket.gethostbyname_ex(socket.gethostname())[2]:
            addrs.add(a)
    except OSError:
        pass
    nets = []
    for a in addrs:
        try:
            ip = ipaddress.IPv4Address(a)
        except ValueError:
            continue
        if ip.is_loopback or ip.is_link_local:
            continue
        nets.append(ipaddress.IPv4Network(f"{a}/24", strict=False))
    return sorted(set(nets), key=str)


def ws_discover(timeout: float = 2.5) -> list[Found]:
    """One ONVIF probe, every answer within `timeout` seconds."""
    out: dict[str, Found] = {}
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
        sock.settimeout(timeout)
        sock.sendto(_PROBE.format(mid=uuid.uuid4()).encode(), _MCAST)
    except OSError:
        return []
    import time
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            data, (host, _) = sock.recvfrom(65535)
        except socket.timeout:
            break
        except OSError:
            break
        f = parse_probe_match(data.decode("utf-8", "replace"), host)
        if f:
            out[f.host] = f
    sock.close()
    return list(out.values())


_XADDR = re.compile(r"<[^>]*XAddrs[^>]*>([^<]+)<")
_SCOPES = re.compile(r"<[^>]*Scopes[^>]*>([^<]+)<")


def parse_probe_match(xml: str, host: str) -> Found | None:
    """Pull the address and the human-readable scopes out of a ProbeMatch.

    A regex rather than an XML parser on purpose: cameras emit every namespace
    prefix imaginable and some emit XML that is not quite well-formed, and the
    two fields wanted are flat text.
    """
    xaddrs = _XADDR.search(xml)
    scopes = _SCOPES.search(xml)
    if not xaddrs and not scopes:
        return None
    url = xaddrs.group(1).split()[0] if xaddrs else ""
    # Prefer the host from the XAddrs URL: a device with several interfaces
    # answers from the one it heard us on, which is the one we can reach.
    m = re.match(r"https?://([^/:]+)", url)
    ip = m.group(1) if m else host
    f = Found(host=ip, onvif=True, onvif_url=url, sources=["onvif"])
    if scopes:
        words = []
        for s in scopes.group(1).split():
            if "/name/" in s or "/hardware/" in s:
                from urllib.parse import unquote
                words.append(unquote(s.rsplit("/", 1)[-1]))
        f.name = " ".join(dict.fromkeys(words))  # dedupe, keep order
    f.make = guess_make(f.name)
    return f


def guess_make(text: str) -> str:
    low = text.lower()
    for needle, make in _MAKES:
        if needle in low:
            return make
    return ""


def rtsp_sweep(nets: Iterable[ipaddress.IPv4Network], timeout: float = 0.5,
               workers: int = 128) -> list[str]:
    """Every host in `nets` with port 554 open. ~254 hosts in about a second."""
    def probe(ip: str) -> str | None:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        try:
            return ip if s.connect_ex((ip, 554)) == 0 else None
        except OSError:
            return None
        finally:
            s.close()
    hosts = [str(h) for n in nets for h in n.hosts()]
    with ThreadPoolExecutor(max_workers=workers) as ex:
        return [ip for ip in ex.map(probe, hosts) if ip]


def discover(timeout: float = 2.5) -> dict:
    """Both probes, merged, as the API returns it."""
    nets = local_networks()
    found: dict[str, Found] = {f.host: f for f in ws_discover(timeout)}
    for ip in rtsp_sweep(nets):
        f = found.setdefault(ip, Found(host=ip))
        f.rtsp = True
        f.sources.append("rtsp")
    cams = sorted(found.values(), key=lambda f: (not (f.rtsp and f.onvif), not f.rtsp,
                                                 ipaddress.IPv4Address(f.host)))
    return {
        "networks": [str(n) for n in nets],
        "cameras": [asdict(c) for c in cams],
    }