ucvl-home-vision/onvif.py

"""Read-only ONVIF Device/Media discovery. Never changes camera configuration."""
import base64
import datetime as dt
import hashlib
import secrets
import urllib.request
import urllib.error
import urllib.parse
import xml.etree.ElementTree as ET
from xml.sax.saxutils import escape

DEVICE = 'http://www.onvif.org/ver10/device/wsdl'
MEDIA = 'http://www.onvif.org/ver10/media/wsdl'
SCHEMA = 'http://www.onvif.org/ver10/schema'


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        # Camera services must stay on the validated device endpoint.
        return None


def check_fault(root):
    fault = root.find('.//{http://www.w3.org/2003/05/soap-envelope}Fault')
    if fault is None:
        return
    codes = [el.text or '' for el in fault.iter() if el.tag.rsplit('}', 1)[-1] in ('Value', 'faultcode')]
    if any(code.rsplit(':', 1)[-1] in ('NotAuthorized', 'FailedAuthentication') for code in codes):
        raise PermissionError('ONVIF authentication failed')
    raise ValueError('ONVIF returned a fault; check device service and request support')


def request(host, port, endpoint, namespace, operation, content, username, password):
    created = dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
    nonce = secrets.token_bytes(16)
    digest = base64.b64encode(hashlib.sha1(nonce + created.encode() + password.encode()).digest()).decode()
    header = ('<wsse:Security s:mustUnderstand="1"><wsse:UsernameToken>'
              '<wsse:Username>' + escape(username) + '</wsse:Username>'
              '<wsse:Password Type="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordDigest">' + digest + '</wsse:Password>'
              '<wsse:Nonce EncodingType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary">' + base64.b64encode(nonce).decode() + '</wsse:Nonce>'
              '<wsu:Created>' + created + '</wsu:Created></wsse:UsernameToken></wsse:Security>')
    envelope = ('<s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope" '
                'xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd" '
                'xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd">'
                '<s:Header>' + header + '</s:Header><s:Body><m:' + operation + ' xmlns:m="' + namespace + '" xmlns:tt="' + SCHEMA + '">'
                + content + '</m:' + operation + '></s:Body></s:Envelope>')
    target = 'http://' + host + ':' + str(port) + endpoint
    passwords = urllib.request.HTTPPasswordMgrWithDefaultRealm()
    passwords.add_password(None, target, username, password)
    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect(), urllib.request.HTTPDigestAuthHandler(passwords))
    req = urllib.request.Request(target, data=envelope.encode(), headers={
        'Content-Type': 'application/soap+xml; charset=utf-8; action="' + namespace + '/' + operation + '"'})
    try:
        with opener.open(req, timeout=5) as response:
            root = ET.fromstring(response.read(1024 * 1024))
    except urllib.error.HTTPError as exc:
        try:
            if exc.code in (401, 403):
                raise PermissionError('ONVIF authentication failed') from None
            try:
                fault = ET.fromstring(exc.read(1024 * 1024))
            except ET.ParseError:
                fault = None
            if fault is not None:
                check_fault(fault)
        finally:
            exc.close()
        raise
    check_fault(root)
    return root


def inspect(host, port, username, password):
    result = {'device': {}, 'profiles': []}
    info = request(host, port, '/onvif/device_service', DEVICE, 'GetDeviceInformation', '', username, password)
    for k in ('Manufacturer', 'Model', 'FirmwareVersion', 'SerialNumber', 'HardwareId'):
        value = info.find('.//{' + DEVICE + '}' + k)
        result['device'][k] = value.text if value is not None else ''
    caps = request(host, port, '/onvif/device_service', DEVICE, 'GetCapabilities', '<m:Category>Media</m:Category>', username, password)
    addr = caps.find('.//{' + SCHEMA + '}Media/{' + SCHEMA + '}XAddr')
    media_endpoint = '/onvif/Media'
    media_port = port
    if addr is not None and addr.text:
        parsed = urllib.parse.urlsplit(addr.text)
        if parsed.hostname == host and parsed.scheme == 'http':
            media_endpoint, media_port = parsed.path, parsed.port or 80
    profiles = request(host, media_port, media_endpoint, MEDIA, 'GetProfiles', '', username, password)
    for profile in profiles.findall('.//{' + MEDIA + '}Profiles')[:32]:
        token = profile.attrib['token']
        def val(name):
            el = profile.find('.//{' + SCHEMA + '}' + name)
            return el.text if el is not None else ''
        content = '<m:StreamSetup><tt:Stream>RTP-Unicast</tt:Stream><tt:Transport><tt:Protocol>RTSP</tt:Protocol></tt:Transport></m:StreamSetup><m:ProfileToken>' + escape(token) + '</m:ProfileToken>'
        response = request(host, media_port, media_endpoint, MEDIA, 'GetStreamUri', content, username, password)
        uri = response.find('.//{' + SCHEMA + '}Uri')
        parsed = urllib.parse.urlsplit(uri.text if uri is not None else '')
        # Only return connection metadata. Never return embedded URL credentials.
        result['profiles'].append({'token': token, 'name': val('Name'), 'sourceToken': val('SourceToken'),
                                   'codec': val('Encoding'), 'width': val('Width'), 'height': val('Height'),
                                   'host': parsed.hostname or host, 'port': parsed.port or 554,
                                   'path': parsed.path + ('?' + parsed.query if parsed.query else '')})
    return result