ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Python实现ONVIF设备自动发现:WS-Discovery多播探测与XAddr验证

Python实现ONVIF设备自动发现:WS-Discovery多播探测与XAddr验证 拿到一个摄像头第一件事永远是“找到它”。不管是做安防平台接入、门禁系统联动还是搞一套智能家居自动化设备发现都是绕不开的第一步。早年大家靠厂家私有SDK一个牌子一套协议集成商最怕的就是客户机房里“万国造”。后来ONVIF协议成了事实标准Open Network Video Interface Forum开放性网络视频接口论坛把IPC、NVR、门禁这类设备统一到了一个标准接口下。但标准接口的“入口”怎么找设备IP是DHCP随机分的用户名密码是厂商默认的总不能一个个网段去盲扫端口。这就是WS-Discovery派上用场的地方。它是ONVIF标准里专门用来做设备自动发现的那一环基于Web Services Dynamic Discovery协议让设备在网络里“自己报名字”或者让你发一个探测包网络里的ONVIF设备必须用自己真正的服务地址回应你。今天这篇就用Python把整套流程走通包括原理、单播探测、多播广播发现以及拿到设备地址后怎么验证这个地址真的能用。全程附完整代码直接照着跑就行。1. 为什么要自己写发现逻辑ONVIF与WS-Discovery的配合关系1.1 ONVIF入门的第一个坎设备地址从哪来ONVIF协议本身是一套基于Web Service的接口规范设备在网络上暴露出来的实际上是一堆SOAP接口地址也就是XAddr。比如某个摄像头的ONVIF媒体服务地址长这样http://192.168.1.64:5080/onvif/media_service云台控制服务是http://192.168.1.64:5080/onvif/ptz_service。你要调ONVIF接口得先拿到这串地址。但问题来了这串地址不是固定的。Dahua的设备可能开在5000端口Hikvision可能开在80端口一台杂牌IPC的ONVIF端口可能是随机高位端口。厂商不同、固件版本不同XAddr的路径也不同。你没法靠猜。所以必须有一个机制让设备主动“告诉你”它的服务地址。ONVIF在设计之初就定下了设备发现协议在Profile S和早期Profile C规范里直接引用了WS-Discovery标准作为发现机制。简单说ONVIF设备会在网络上宣告自己“我是ONVIF设备我的类型是NetworkVideoTransmitter我在这里”或者在收到探测请求后把自己的XAddr回复给对方。这也是我们能写自动化脚本做网络扫描的理论依据。1.2 WS-Discovery的工作模式与ONVIF的约定WS-Discovery协议定义了四种消息类型Hello、Bye、Probe、ProbeMatch。其中Hello是设备上线时主动宣告Bye是设备下线时通告Probe是客户端发起的能力探测ProbeMatch则是设备对Probe的回复。放到ONVIF场景里我们最常用的是Probe和ProbeMatch。客户端向一个固定的多播地址和端口——239.255.255.250:3702——发送一段XML格式的SOAP探测消息询问“谁是ONVIF设备”。网络里的ONVIF设备收到这个消息后如果没有特殊配置拦截就会回复一个ProbeMatch消息告诉客户端“我是NVT类型设备我的地址是xxx请往这里来”。还有一点值得注意WS-Discovery虽然是多播协议但它同样支持单播探测。也就是说你甚至可以往某个已知IP的3702端口单独发Probe只问这台设备。这个特性对我们做定向扫描非常有用后面代码里我会把两种方式都实现一遍。2. 核心原理拆解一次Probe探测的完整生命周期2.1 为什么能收到响应UDP多播的运作机制很多人第一次接触多播时会有个错觉觉得往广播地址发个包就能收到全网设备的回复。实际上多播和广播是有本质区别的。广播是“送到网段里每一台主机”多播是“只送给加了那个多播组的主机”。WS-Discovery用的239.255.255.250属于IPv4本地网段管理多播地址路由器默认不会转发这个地址范围的包只在当前二层网络内有效。设备如果实现了ONVIF发现服务它会在系统启动时加入这个多播组所以当探测包到达时操作系统会把包交给ONVIF的应用层处理。关键点在这里设备响应Probe时不一定是通过多播回包。很多设备是拿到你发送方的IP和端口后用单播直接回复。所以你在写代码时不能只监听多播地址必须同时监听发送时绑定的那个IP和端口。代码里我们要把UDP socket绑定到0.0.0.0:3702这样无论是发到多播组的回包还是直接单播回包都能收到。2.2 抓包视角看一次完整的探测交互我在本地用Wireshark抓过一次包整个交互非常清晰。客户端构造出这样一段XML作为UDP载荷发出去?xml version1.0 encodingUTF-8? e:Envelope xmlns:ehttp://www.w3.org/2003/05/soap-envelope xmlns:whttp://schemas.xmlsoap.org/ws/2004/08/addressing xmlns:dhttp://schemas.xmlsoap.org/ws/2005/04/discovery xmlns:dnhttp://www.onvif.org/ver10/network/wsdl e:Header w:MessageIDuuid:9af7d2c0-5b6e-4b1c-8a1e-6b96b4d74a91/w:MessageID w:Tourn:schemas-xmlsoap-org:ws:2005:04:discovery/w:To w:Actionhttp://schemas.xmlsoap.org/ws/2005/04/discovery/Probe/w:Action /e:Header e:Body d:Probe d:Typesdn:NetworkVideoTransmitter/d:Types /d:Probe /e:Body /e:Envelope这段XML就是整个探测的核心。MessageID是UUID用于匹配请求和响应。Types字段是关键dn:NetworkVideoTransmitter表示“我要找网络视频发送设备”这是ONVIF对IP摄像机的标准类型定义。有些设备固件实现不严格你发这个Types它不响应可以改成空Probe去掉Types标签就能收到响应。设备收到后回包大概长这样e:Envelope xmlns:ehttp://www.w3.org/2003/05/soap-envelope ... e:Header w:RelatesTouuid:9af7d2c0-5b6e-4b1c-8a1e-6b96b4d74a91/w:RelatesTo ... /e:Header e:Body d:ProbeMatches d:ProbeMatch w:Addresshttp://192.168.1.64:5080/onvif/device_service/w:Address d:Typesdn:NetworkVideoTransmitter/d:Types d:Scopesonvif://www.onvif.org/type/video_encoder .../d:Scopes d:XAddrshttp://192.168.1.64:5080/onvif/device_service/d:XAddrs /d:ProbeMatch /d:ProbeMatches /e:Body /e:EnvelopeAddress是设备的WS-Addressing地址XAddrs是实际的ONVIF服务地址列表这个才是我们要的。Scopes里有设备的硬件ID、MAC地址、名字等元信息解析出来可以在扫描结果里直接显示设备名。2.3 两种探测方式的取舍与适用场景我这里把单播探测和多播发现都列了出来实际项目里怎么选取决于你的场景。单播探测适合“知道IP但不知道ONVIF接口路径”的情况。比如你扫描到网段内所有开放端口的IP想逐一确认哪个是ONVIF设备。它的优点是快、准、不用加入多播组缺点是得先知道目标IP没法主动发现“还有哪些设备存在”。多播发现适合“一无所知纯粹想知道网络里有什么设备”的情况。比如你接到一个新项目整个机房的摄像头都在一个二层网络里你连IP网段都没记录。这时发一个多播Probe所有同网段的ONVIF设备都会回复一份清单就出来了。缺点是多播包只在本网段内传播跨三层就得靠代理或中继设计不过绝大多数安防项目都在同一内网够用了。3. 完整代码实现从单播探测到多播自动发现3.1 环境准备与依赖说明代码只用Python标准库不需要安装任何第三方包Python 3.8以上即可运行。如果你用的是较老的2.7版本语法需要微调但我建议直接上3.x后面解析XML的部分能省不少事。如果你是第一次跑Python脚本先确认命令行里能执行python3 --version。Windows用户可能要用py -3 --version这取决于你安装时的PATH配置。没装Python的话去官网下载安装包安装时一定记得勾选“Add Python to PATH”复选框否则后面在cmd里摸不到python命令。代码我放在一个文件里拆成了两个函数一个是probe_single(ip)定向探测单个IP一个是probe_multicast(timeout)全网段多播发现。最后再加一个fetch_xaddr(xaddr)用HTTP请求验证发现的XAddr是否真实可用。3.2 单播探测快速定向识别ONVIF设备先看单播探测。这个函数的核心就三步构造SOAP包、UDP发出去、监听回复。import socket import uuid import xml.etree.ElementTree as ET ONVIF_PORT 3702 WS_DISCOVERY_ADDR 239.255.255.250 def build_probe_xml(): message_id uuid: str(uuid.uuid4()) return f?xml version1.0 encodingUTF-8? e:Envelope xmlns:ehttp://www.w3.org/2003/05/soap-envelope xmlns:whttp://schemas.xmlsoap.org/ws/2004/08/addressing xmlns:dhttp://schemas.xmlsoap.org/ws/2005/04/discovery xmlns:dnhttp://www.onvif.org/ver10/network/wsdl e:Header w:MessageID{message_id}/w:MessageID w:Tourn:schemas-xmlsoap-org:ws:2005:04:discovery/w:To w:Actionhttp://schemas.xmlsoap.org/ws/2005/04/discovery/Probe/w:Action /e:Header e:Body d:Probe d:Typesdn:NetworkVideoTransmitter/d:Types /d:Probe /e:Body /e:Envelope def probe_single(ip, timeout3): sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(timeout) sock.bind((0.0.0.0, 0)) probe_xml build_probe_xml().encode(utf-8) sock.sendto(probe_xml, (ip, ONVIF_PORT)) print(f[*] 已向 {ip}:{ONVIF_PORT} 发送单播探测包) try: data, addr sock.recvfrom(65535) print(f[] 收到来自 {addr[0]}:{addr[1]} 的响应) parse_probe_response(data.decode(utf-8, errorsignore)) except socket.timeout: print(f[-] {ip} 在 {timeout} 秒内没有响应可能不是ONVIF设备或端口被过滤) finally: sock.close()注意几个细节。sock.bind((0.0.0.0, 0))这行端口0表示让操作系统随机分配一个可用端口。这样设备回包时能直接找到我们。UDP是面向无连接的sendto之后立刻调用recvfrom阻塞等待。局域网内的ONVIF设备一般几十毫秒内就会响应设置3秒超时已经足够。3.3 多播发现一张包扫出全网段设备单播探测虽然简单但你要是一台一台去试效率太低了。多播发现才是自动化的精髓。Python里发多播包需要几步特殊处理设置TTL、加入多播组、设置IP_MULTICAST_LOOP。import struct def probe_multicast(timeout5): sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((0.0.0.0, ONVIF_PORT)) # 设置多播TTL为2防止跨路由器转发 sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2) # 允许接收本机发出的多播包 sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1) # 加入多播组 mreq struct.pack(4sl, socket.inet_aton(WS_DISCOVERY_ADDR), socket.INADDR_ANY) sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) sock.settimeout(timeout) probe_xml build_probe_xml().encode(utf-8) sent sock.sendto(probe_xml, (WS_DISCOVERY_ADDR, ONVIF_PORT)) print(f[*] 已向 {WS_DISCOVERY_ADDR}:{ONVIF_PORT} 发送多播探测包共 {sent} 字节) devices [] try: while True: data, addr sock.recvfrom(65535) info parse_probe_response(data.decode(utf-8, errorsignore)) if info: info[source_ip] addr[0] devices.append(info) except socket.timeout: print(f[*] 等待 {timeout} 秒结束共发现 {len(devices)} 台设备) finally: sock.close() return devices关键点逐一说明。SO_REUSEADDR必须设置否则脚本连续运行时端口可能因为TIME_WAIT状态无法复用报“Address already in use”的错误。IP_MULTICAST_TTL设成2就够了既能在本网段正常通信又不怕误发到外部网络。还有最容易被忽略的IP_ADD_MEMBERSHIP不仅是发送方需要加入多播组接收方要是没加入就算包发到主机网卡上操作系统也不会把数据送到你的socket。3.4 解析响应用ElementTree提取XAddr前面两个函数都用到了parse_probe_response现在把它的实现贴出来。WS-Discovery响应是SOAP XML格式包含命名空间用正则匹配会很痛苦用ElementTree处理则很顺手。def parse_probe_response(xml_text): ns { e: http://www.w3.org/2003/05/soap-envelope, w: http://schemas.xmlsoap.org/ws/2004/08/addressing, d: http://schemas.xmlsoap.org/ws/2005/04/discovery, dn: http://www.onvif.org/ver10/network/wsdl, } try: root ET.fromstring(xml_text) probes root.findall(.//d:ProbeMatches/d:ProbeMatch, ns) if not probes: return None result [] for probe in probes: addr probe.find(w:Address, ns) xaddrs probe.find(d:XAddrs, ns) types probe.find(d:Types, ns) scopes probe.find(d:Scopes, ns) item { address: addr.text if addr is not None else , xaddrs: xaddrs.text if xaddrs is not None else , types: types.text if types is not None else , scopes: scopes.text if scopes is not None else , } result.append(item) return result except ET.ParseError as ex: print(f[!] XML解析失败: {ex}) return None这里我在单播分支里调用parse_probe_response后没有继续处理返回值但在多播分支里要收集结果所以函数统一返回列表。你实际自己改造时单播分支也可以把返回值接住打印出详细内容。还有一个细节有些设备会返回多个ProbeMatch比如一个物理设备上有多个逻辑实体所以这里用了findall而不是find。3.5 验证XAddr发出的HTTP请求确认不是“死链”WS-Discovery拿到了XAddr但别高兴太早。我在实际项目里遇到过一种情况设备回复的XAddr是一个对外不可达的地址比如设备在多网卡环境下把管理口和视频口搞混了。为了不把脏数据带进系统最好对XAddr做一个HTTP GET验证。ONVIF设备服务对GET请求通常会返回SOAP错误或HTTP 200但只要是有效HTTP服务且没有连接超时基本可以判定这个地址是真实的。import urllib.request import urllib.error def verify_xaddr(xaddr, timeout3): req urllib.request.Request(xaddr, methodGET) try: resp urllib.request.urlopen(req, timeouttimeout) print(f[] HTTP {resp.status}XAddr可访问: {xaddr}) return True except urllib.error.HTTPError as ex: # 2xx之外的状态码不一定代表失败SOAP服务对GET经常会回405或500 print(f[*] HTTP {ex.code}但服务存在ONVIF通常对GET返回错误: {xaddr}) return True except Exception as ex: print(f[-] XAddr不可访问: {xaddr}原因: {ex}) return False注意一个容易误判的点ONVIF的Web服务接口如果只用GET访问不带SOAPAction头多数会返回HTTP 400、405甚至500。这不代表服务不存在恰恰说明服务端是活的。所以只要不是URLError超时我都算它是有效地址。真正的死链通常是连接超时或DNS解析失败这两种才需要标记为不可用。4. 常见问题与排查技巧实录4.1 多播探测器收不到任何设备响应这是所有人第一次跑脚本时几乎必踩的坑。我在实验室里模拟过好几回总结出以下几个高频原因。第一脚本运行环境和设备不在同一个二层网络。多播不会跨路由转发如果你的电脑连着Wi-Fi摄像头插在另一个交换机下面即使同一个网段也可能因为AP的多播隔离策略收不到包。用网线直连设备或让电脑和设备接入同一台交换机问题立刻解决。第二Windows防火墙拦截了来自UDP 3702端口的入站流量。临时关掉防火墙测试是最快的验证方法确认是防火墙问题后再去高级安全里把3702端口或“专用网络”的入站UDP规则放开。第三代码没有绑定正确的IP。如果电脑有多个网卡Wi-Fi、有线、虚拟网卡socket默认绑定的接口可能不对。这时需要给socket显式指定发送网卡也就是在IP_MULTICAST_IF里设置接口IP。下面这段可以加在join组之前# 指定发送多播包的网卡IP按实际环境改 sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(192.168.1.100))4.2 收到了响应但是XAddr连不上这种情况多半是设备回包给了内网IP而你的电脑访问不了那个IP。我遇到过Dahua的设备在多网卡模式下回包给出了另一个网段的地址导致XAddr无法访问。还有一种是设备开启了“多播发现隔离”或“跨网段发现限制”的加密选项回包内容会被mask掉一部分XAddr变成0.0.0.0。这种就只能登录设备后台关掉相应的限制策略。如果是跨网段场景在路由设备上添加一条到摄像头网段的路由或者直接把脚本部署到摄像头所在的同网段机器上跑最省事。4.3 部分设备能发现但类型过滤后消失dn:NetworkVideoTransmitter是ONVIF标准里对网络视频发射端Network Video TransmitterNVT的定义绝大多数IPC会主动匹配这个类型。但有些设备固件实现得比较草率在d:Types字段里写得和标准不完全一致甚至直接把Types字段留空。遇到这种情况把Probe消息里的d:Types标签整个去掉做成“来者不拒”的探测方式响应率会高很多。我通常会把“严格匹配NVT类型”和“宽松探测全部类型”做成两个开关先跑一次严格的拿到标准设备清单再跑一次宽松的看看有没有漏网之鱼。两者结果做差集就能发现某些不太守规矩的设备。下面这个变更只需要改一行def build_probe_xml(strictTrue): # ... if strict: probe_type d:Typesdn:NetworkVideoTransmitter/d:Types else: probe_type # ...4.4 解析响应时报XML格式错误偶尔某些设备的SOAP响应里会带一些不可见字符或者头部的XML声明编码格式不规范。用errorsignore能规避一部分非UTF-8编码的问题但如果设备返回的报文被截断成半个XMLElementTree照样会报ParseError。这类情况建议直接扔掉这条响应不影响整体发现结果。另外在解析之前别忘了一件事WS-Discovery的响应是一个完整UDP报文但网络层并不会保证你的recvfrom一次能收全。虽然绝大多数ONVIF设备的ProbeMatch响应小于1500字节不会触发UDP分片但如果你在跨度很大的WAN环境跑脚本还是建议对收到的buffer长度做一次判断超过1400字节的再检查一下是否需要重组。这个概率极低但排查到的时候会让人很崩溃。5. 完整演示从扫描到确认一条龙最后把三个功能串起来写一个真正能落地的main入口。这个入口接收命令行参数单个IP或网段对应的做单播探测或多播发现然后自动验证每个XAddr最后按表格格式输出。import sys import time def main(): if len(sys.argv) 1 and sys.argv[1] --single: ip sys.argv[2] probe_single(ip) return devices probe_multicast(timeout5) if not devices: print(没有发现任何ONVIF设备) return # 这里对扁平化后的每个XAddr做验证 verified_count 0 print(\n 发现结果 ) for dev_group in devices: # parse_probe_response返回的是list这里兼容一下 for item in dev_group: print(f来源IP: {item.get(source_ip, N/A)}) print(f Address: {item.get(address)}) print(f XAddrs: {item.get(xaddrs)}) xaddr (item.get(xaddrs) or ).strip() if xaddr and verify_xaddr(xaddr.split()[0]): verified_count 1 print(- * 40) print(f共发现 {len(devices)} 组设备其中 {verified_count} 组验证可用)等一下parse_probe_response在多播分支的调用里返回值是一个列表但在上面的代码示例里我直接把devices里的元素打印成功实际需要注意类型。严谨一点的做法是在probe_multicast中解析函数返回的是列表时用extend而不是appendif info: devices.extend(info)这样后面遍历时每个元素就是一个字典取字段时可直接用。这个细节很容易忽略但我见过好几个人在这卡住。最终完整代码我放在文末GitHub Gist链接里跑之前确保把第4节提到的几个环境问题排查一遍。6. 拿到XAddr之后能做什么设备发现只是入场券。XAddr拿到手意味着你拿到了进入这个设备ONVIF服务大门的钥匙。后续做的事五花八门用device_service地址做GetSystemDateAndTime、GetDeviceInformation读取设备型号和固件版本。用media_service地址做GetProfiles、GetStreamUri拿到RTSP拉流地址把视频接到自己的播放器或AI分析框架里。用ptz_service地址控制云台转动、预置位调用。这些都是标准化接口不管什么品牌的设备只要通过了ONVIF认证调用逻辑全都一样。如果你要做的只是快速验证设备是否存在也可以用现成的工具比如ONVIF Device ManagerODM这个神器图形界面上点几下就能看到设备信息和取流地址。但它不能替代你自己的脚本。ODM能做的只是人机交互没法嵌入到你的监控平台、自动巡检程序或设备资产管理系统里。我经常干的一件事是办公室资产盘点时写一个定时任务定期多播探测一次自动生成全楼摄像头在线清单哪个摄像头掉线了脚本一跑十分钟内就能发现。根据我的实际排查经验最稳定的设备发现策略其实是多播探测为主、单播扫描为辅、HTTP验证兜底这个组合能覆盖九成以上的安防网络环境。设备发现写完之后把结果存进SQLite或Excel配合企业微信或钉钉机器人做告警就是一套很实用的设备状态监控方案了。
返回列表