杰瑞科技汇

Python如何实现WS-Discovery协议?

Of course! Using Python for WS-Discovery is a common task for finding network devices like printers, scanners, cameras, and other industrial equipment that implement the Web Services on Devices (WS-Dev) profile.

Python如何实现WS-Discovery协议?-图1
(图片来源网络,侵删)

Here’s a comprehensive guide covering the concepts, popular libraries, and practical code examples.


What is WS-Discovery?

WS-Discovery (Web Services Discovery) is a technical standard by Microsoft (and now OASIS) that defines a multicast discovery protocol. It allows a client application to dynamically discover the presence and network location of services on a local network.

Think of it as the network equivalent of shouting "Is anyone there?" and all the compatible devices shouting back "I'm here, and this is what I can do!"

Key Concepts:

Python如何实现WS-Discovery协议?-图2
(图片来源网络,侵删)
  • Multicast: The client sends a request to a special IP address (255.255.250) and port (3702) that all devices on the local subnet listen to. This is efficient as it doesn't require knowing the IP addresses of potential devices beforehand.
  • Hello/Bye Messages: Devices periodically send "Hello" messages to announce their presence. When they shut down, they send "Bye" messages.
  • Probe/Match Messages: A client sends a "Probe" message asking for devices with specific capabilities. Devices that match the criteria respond with a "Match" message.
  • SOAP over UDP: The messages are formatted as SOAP (Simple Object Access Protocol) packets and sent over UDP (User Datagram Protocol), making it lightweight and fast.

Python Libraries for WS-Discovery

There are a few excellent libraries to choose from, depending on your needs.

Library Description Pros Cons
python-ws-discovery A pure Python, lightweight library focused on being a client and a server. Easy to use, pure Python (no external dependencies), good for simple discovery. Less actively maintained, may not support the latest WS-* standards.
zeroconf The Python implementation of mDNS/DNS-SD (Multicast DNS / DNS Service Discovery). Highly recommended. Very powerful, actively maintained, used by millions (Bonjour/Avahi). It's the de facto standard for service discovery on local networks. It discovers services by type (e.g., _ipp._tcp.local.), not by WS-Discovery's SOAP-based filtering. You get a list of IPs and ports, then you need to query the device for its capabilities.
zeep A modern, feature-rich SOAP client library. Can act as a WS-Discovery client by manually building and sending SOAP messages. Extremely powerful for interacting with the discovered SOAP services. More complex setup, requires understanding SOAP message structure. Not a dedicated discovery library.

Recommendation:

  • For most use cases, start with zeroconf. It's the most robust and community-supported solution for finding network services.
  • If you need to filter devices based on their specific WS-Discovery Types or Scopes (e.g., only find printers), and you want a simpler API, python-ws-discovery is a good choice.
  • If you already use zeep for interacting with web services, you can use it for discovery as well.

Practical Examples

Example 1: Using zeroconf (Recommended)

This is the most common and practical approach. You'll discover devices by their service type and then inspect the information they provide.

Installation:

Python如何实现WS-Discovery协议?-图3
(图片来源网络,侵删)
pip install zeroconf

Code: This script will discover all _ipp._tcp (Internet Printing Protocol) services on the network, which is common for network printers.

import time
from zeroconf import ServiceBrowser, ServiceListener, Zeroconf
class MyListener(ServiceListener):
    """A listener for zeroconf services."""
    def add_service(self, zeroconf: Zeroconf, type: str, name: str) -> None:
        print(f"Found service: {name} of type {type}")
        # Get the service info
        info = zeroconf.get_service_info(type, name)
        if info:
            print(f"  - Address: {info.addresses}")
            print(f"  - Port: {info.port}")
            print(f"  - Properties: {info.properties}")
            print("-" * 50)
            # You can now use the address and port to connect to the service
            # For a printer, you might connect to the port for a raw socket or HTTP.
            # e.g., http://{info.addresses[0]}:{info.port}/
    def remove_service(self, zeroconf: Zeroconf, type: str, name: str) -> None:
        print(f"Service removed: {name}")
    def update_service(self, zeroconf: Zeroconf, type: str, name: str) -> None:
        print(f"Service updated: {name}")
# The service type for IPP (commonly used by printers)
SERVICE_TYPE = "_ipp._tcp.local."
# Create a Zeroconf instance
zc = Zeroconf()
listener = MyListener()
# Start browsing for services of the given type
print(f"Browsing for {SERVICE_TYPE} services...")
browser = ServiceBrowser(zc, SERVICE_TYPE, listener)
try:
    # Keep the script running to listen for messages
    while True:
        time.sleep(0.1)
except KeyboardInterrupt:
    pass
finally:
    print("Stopping browser...")
    browser.cancel()
    zc.close()
    print("Done.")

Example 2: Using python-ws-discovery

This library is more focused on the WS-Discovery protocol itself, allowing you to send Probe messages with specific criteria.

Installation:

pip install python-ws-discovery

Code: This script will actively probe for any WS-Discovery device on the network and print the details of the responses.

import logging
from wsdiscovery import WSDiscovery
# Set up logging to see the messages being sent and received
logging.basicConfig(level=logging.INFO)
# Create a WSDiscovery instance. This starts a listener in the background.
# You don't need to call `search()` to listen for Hello/Bye messages.
wsd = WSDiscovery()
# Start the discovery service (listener and sender)
wsd.start()
print("Discovering WS-Discovery services... Press Ctrl+C to stop.")
print("Listening for Hello and Bye messages...")
try:
    # Let it run for a while to catch periodic "Hello" messages
    # and any "Probe" responses if another client is searching.
    while True:
        time.sleep(1)
        # You can also trigger an active probe
        # wsd.search(types=['wsd:Device'])
except KeyboardInterrupt:
    pass
finally:
    print("Stopping discovery service...")
    wsd.stop()
    print("Done.")

To actively search for devices with a specific capability (e.g., printers):

from wsdiscovery import WSDiscovery
import time
wsd = WSDiscovery()
wsd.start()
# Search for devices that are printers
# The 'types' parameter is a list of QNames (Qualified Names)
# Common types for printers are 'wsd:Device' or specific manufacturer types.
print("Actively probing for printer devices...")
wsd.search(types=['wsd:Device'])
# Give some time for devices to respond
time.sleep(5)
# Get the list of discovered services
services = wsd.getServices()
if services:
    print(f"\nFound {len(services)} services:")
    for service in services:
        print(f"  - Endpoint: {service.getXAddrs()}")
        print(f"    Types: {service.getTypes()}")
        print(f"    Scopes: {service.getScopes()}")
        print(f"    Metadata: {service.getMetadata()}")
        print("-" * 20)
else:
    print("No services found.")
wsd.stop()

Advanced: Using zeep for Manual Discovery

This is the most "hardcore" approach. You manually construct the SOAP Probe message and send it via UDP. This is useful if you need fine-grained control or are working with a non-standard implementation.

Installation:

pip install zeep

Code: This example uses zeep to create the SOAP message and socket to send it via UDP.

import socket
import struct
import time
from zeep import xsd
# Define the SOAP message structure for a WS-Discovery Probe
# This is based on the WS-Discovery specification
probe_schema = xsd.Schema(
    """
    <xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema"
                xmlns:wsd="http://schemas.xmlsoap.org/ws/2005/04/discovery"
                targetNamespace="http://schemas.xmlsoap.org/ws/2005/04/discovery">
      <xsd:element name="Probe">
        <xsd:complexType>
          <xsd:sequence>
            <xsd:element name="Types" type="xsd:QName" minOccurs="0" maxOccurs="1"/>
          </xsd:sequence>
        </xsd:complexType>
      </xsd:element>
    </xsd:schema>
    """
)
# Create the Probe message
probe_message = probe_schema.get_element('Probe')(
    # To find all devices, omit the Types element.
    # To find specific devices, add the QName, e.g., xsd.QName('http://schemas.xmlsoap.org/ws/2005/04/discovery', 'Device')
    Types=xsd.QName('http://schemas.xmlsoap.org/ws/2005/04/discovery', 'Device')
)
# Serialize to XML
soap_envelope = f"""
<soap:Envelope xmlns:soap="http://www.w3.org/2003/05/soap-envelope"
               xmlns:wsd="http://schemas.xmlsoap.org/ws/2005/04/discovery">
  <soap:Header/>
  <soap:Body>
    {probe_message}
  </soap:Body>
</soap:Envelope>
"""
# WS-Discovery specific headers and UDP packet structure
uuid = "uuid:1b3a5c7d-9e1f-4g0h-i2j3-k4l5m6n7o8p9" # A unique ID for this request
msg_id = f"urn:uuid:{uuid}"
to_addr = "urn:schemas-xmlsoap-org:ws:2005:04:addressing:role:anonymous"
action = "http://schemas.xmlsoap.org/ws/2005/04/discovery/Probe"
# Create the SOAP message with WS-Discovery headers
full_message = f"""<s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope">
<s:Header>
    <a:Action s:mustUnderstand="1" xmlns:a="http://schemas.xmlsoap.org/ws/2005/04/addressing">{action}</a:Action>
    <a:MessageID>{msg_id}</a:MessageID>
    <a:To s:mustUnderstand="1">{to_addr}</a:To>
    <d:Probe xmlns:d="http://schemas.xmlsoap.org/ws/2005/04/discovery">
        <d:Types>wsd:Device</d:Types>
    </d:Probe>
</s:Header>
<s:Body/>
</s:Envelope>"""
# WS-Discovery uses a specific UDP packet format
# This is a simplified version. A full implementation would include
# SOAPAction header and other WS-* addressing headers.
# The message must be wrapped in a <SOAP-ENV:Envelope> tag.
final_message = full_message.encode('utf-8')
# Create a UDP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.settimeout(5) # Wait 5 seconds for a response
# Send the message to the WS-Discovery multicast address
multicast_group = ('239.255.255.250', 3702)
print(f"Sending Probe to {multicast_group}...")
sock.sendto(final_message, multicast_group)
try:
    while True:
        # Receive a response
        data, addr = sock.recvfrom(65535)
        print(f"\nReceived response from {addr[0]}:{addr[1]}")
        print("Raw SOAP Response:")
        print(data.decode('utf-8'))
        # In a real application, you would parse this XML with zeep
        # to find the XAddrs (endpoint addresses) of the discovered device.
except socket.timeout:
    print("No more responses received.")
finally:
    sock.close()

Summary and Best Practices

  1. Start with zeroconf: It's the most reliable and widely used method for local network service discovery. It's simple and effective.
  2. Use python-ws-discovery for Filtering: If you need to filter your search based on WS-Discovery specific Types or Scopes and want a simpler API than manually crafting SOAP messages, this is a great choice.
  3. Parse the Response: Once you discover a device, the response (whether from zeroconf or a raw SOAP message) will contain its network address (XAddrs) and metadata. Use this information to connect to the device's actual service (e.g., its web interface, a TCP port for data, etc.).
  4. Handle Timeouts and Errors: Network discovery is not 100% reliable. Always include timeouts and error handling in your code. Devices might be busy or temporarily unavailable.
分享:
扫描分享到社交APP
上一篇
下一篇