Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def announce(ip_address, port):
"""Announces that *Virtual Touchpad* is available on the local network.
:param str ip_address: The IP address on which *Virtual Touchpad* is reachable.
:param int port: The port on which to connect to *Virtual Touchpad*.
:return: an object with the method ``unregister()``, which must be called to
remove the service from the system
"""
import getpass
import socket
import types
result = zeroconf.Zeroconf()
info = zeroconf.ServiceInfo(
SERVICE_NAME,
'%s@%s.%s' % (getpass.getuser(), socket.gethostname(), SERVICE_NAME),
socket.inet_aton(ip_address),
port,
0, 0, # weight, priority
{
'version': '.'.join(str(v) for v in __version__)})
result.register_service(info)
def unregister(self):
"""Unregisters the Virtual Touchpad service.
"""
self.unregister_service(info)
self.close()
result.unregister = types.MethodType(unregister, result)
def _server_register(self):
with self._zconf_lock:
hostname = socket.gethostname()
try:
ips = socket.gethostbyname_ex(hostname)[-1]
except:
ips = None
if ips:
compname = self._get_server_name().decode('utf-8')
hostaddr = ips[0]
desc = {'platform': self._SERVER_PLATFORM, 'version': '%s.%s.%s'%self._SERVER_VERSION}
info = zeroconf.ServiceInfo(self._ZEROCONF_TYPE, '%s.'%compname + self._ZEROCONF_TYPE, socket.inet_aton(hostaddr), self._tcp_srv_port, 0, 0, desc, '%s.'%hostname)
self._zconf = zeroconf.Zeroconf(bindaddress=hostaddr)
self._zconf.register_service(info)
else:
self._zconf = zeroconf.Zeroconf()
# Other useful attributes for the connection.
desc = {
'id': self.SERVICE_NAME,
'version': __version__,
'os': CURRENT_PLATFORM.name,
'api': self._api_name
}
# The name can't have '.', because it's a special character used as
# a separator, and some NSD clients can't handle names with it.
system = f"{platform.system()} {platform.node()}".replace('.', '_')
full_name = f"{self.SERVICE_NAME} - {system}"
# The name's maximum length is 64 bytes
if len(full_name) >= 64:
full_name = full_name[:60] + "..."
self.info = ServiceInfo(
self.SERVICE_TYPE,
f"{full_name}.{self.SERVICE_TYPE}",
addresses=[socket.inet_aton(self.address)],
port=self.port,
properties=desc)
logging.info("Registering Vidify service")
self.zeroconf = Zeroconf(ip_version=self.IP_VERSION)
self.zeroconf.register_service(self.info)
def _init_service(self):
properties = self.get_properties()
server = self.config.host_name + '.local.'
service_name = self.config.service_id + '._airdrop._tcp.local.'
info = ServiceInfo('_airdrop._tcp.local.',
service_name,
port=self.config.port,
properties=properties,
server=server,
addresses=[self.ip_addr.packed])
return info
self.unregisterAllServices()
self.socket.setsockopt(socket.SOL_IP,
socket.IP_DROP_MEMBERSHIP,
socket.inet_aton(_MDNS_ADDR) +
socket.inet_aton('0.0.0.0'))
self.socket.close()
# Test a few module features, including service registration, service
# query (for Zoe), and service unregistration.
if __name__ == '__main__':
print "Multicast DNS Service Discovery for Python, version", __version__
r = Zeroconf()
print "1. Testing registration of a service..."
desc = {'version':'0.10','a':'test value', 'b':'another value'}
info = ServiceInfo("_http._tcp.local.",
"My Service Name._http._tcp.local.",
socket.inet_aton("127.0.0.1"), 1234, 0, 0, desc)
print " Registering service..."
r.registerService(info)
print " Registration done."
print "2. Testing query of service information..."
print " Getting ZOE service:",
print str(r.getServiceInfo("_http._tcp.local.", "ZOE._http._tcp.local."))
print " Query done."
print "3. Testing query of own service..."
print " Getting self:",
print str(r.getServiceInfo("_http._tcp.local.",
"My Service Name._http._tcp.local."))
print " Query done."
print "4. Testing unregister of service information..."
r.unregisterService(info)
def _onServiceChanged(self, zero_conf, service_type, name, state_change):
if state_change == ServiceStateChange.Added:
# First try getting info from zero-conf cache
info = ServiceInfo(service_type, name, properties = {})
for record in zero_conf.cache.entries_with_name(name.lower()):
info.update_record(zero_conf, time(), record)
for record in zero_conf.cache.entries_with_name(info.server):
info.update_record(zero_conf, time(), record)
if info.address:
break
# Request more data if info is not complete
if not info.address:
info = zero_conf.get_service_info(service_type, name)
if info:
type_of_device = info.properties.get(b"type", None)
if type_of_device:
if type_of_device == b"printer":
else:
machine_id = hex(generate_persistent_id())
database_id = hex(generate_persistent_id())
# iTunes 11+ uses more properties, but this seems to be sufficient.
description = {
"txtvers": "1",
"Password": str(int(bool(daap_server.password))),
"Machine Name": provider.server.name,
"Machine ID": machine_id.upper(),
"Database ID": database_id.upper()
}
# Test is zeroconf supports multiple addresses or not. For
# compatibility with zeroconf 0.17.3 or less.
if not hasattr(zeroconf.ServiceInfo("", ""), "addresses"):
addresses = addresses[0]
self.daap_servers[daap_server] = zeroconf.ServiceInfo(
type="_daap._tcp.local.",
name=provider.server.name + "._daap._tcp.local.",
address=addresses,
port=daap_server.port,
properties=description)
self.zeroconf.register_service(self.daap_servers[daap_server])
reachable.
:param int port: The port on which to connect to *Virtual Touchpad*.
"""
import getpass
import socket
import types
try:
import zeroconf
except:
yield
return
zc = zeroconf.Zeroconf()
info = zeroconf.ServiceInfo(
SERVICE_NAME,
'%s@%s.%s' % (
getpass.getuser(),
socket.gethostname().replace('.', '_'),
SERVICE_NAME),
socket.inet_aton(ip_address),
port,
0, 0,
{
'version': '.'.join(str(v) for v in __version__)})
zc.register_service(info)
try:
yield
finally:
zc.unregister_service(info)