Codex
84 lines
5.1 KiB
Python
84 lines
5.1 KiB
Python
"""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 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:
|
|
exc.close()
|
|
raise
|
|
if root.find('.//{http://www.w3.org/2003/05/soap-envelope}Fault') is not None:
|
|
raise ValueError('ONVIF returned a fault; check device credentials and permissions')
|
|
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
|