Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"OPT: \"http://schemas.upnp.org/upnp/1/0/\"; ns=01\r\n"
"01-NLS: %s\r\n"
"SERVER: %s\r\n"
"ST: %s\r\n"
"USN: uuid:%s::%s\r\n" % (date_str, location_url, self.uuid, self.server_version, search_target, self.persistent_uuid, search_target))
if self.other_headers:
for header in self.other_headers:
message += "%s\r\n" % header
message += "\r\n"
temp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
temp_socket.sendto(bytes(message, 'UTF-8'), destination)
#print("Responding to search-->" + message )
# This subclass does the bulk of the work to mimic a WeMo switch on the network.
class fauxmo(upnp_device):
@staticmethod
def make_uuid(name):
return ''.join(["%x" % sum([ord(c) for c in name])] + ["%x" % ord(c) for c in "%sfauxmo!" % name])[:14]
def __init__(self, name, listener, poller, ip_address, port, action_handler = None, log = None):
self.serial = self.make_uuid(name)
self.name = name
self.switchStatus=0
self.ip_address = ip_address
if (log != None):
self.log = log
persistent_uuid = "Socket-1_0-" + self.serial
other_headers = ['X-User-Agent: redsonic']
upnp_device.__init__(self, listener, poller, port, "http://%(ip_address)s:%(port)s/setup.xml", "Unspecified, UPnP/1.0, Unspecified", persistent_uuid, other_headers=other_headers, ip_address=ip_address, log=self.log)
if action_handler:
self.action_handler = action_handler
def __init__(self, listener, poller, port, root_url, server_version, persistent_uuid, other_headers = None, ip_address = None):
self.listener = listener
self.poller = poller
self.port = port
self.root_url = root_url
self.server_version = server_version
self.persistent_uuid = persistent_uuid
self.uuid = uuid.uuid4()
self.other_headers = other_headers
if ip_address:
self.ip_address = ip_address
else:
self.ip_address = upnp_device.local_ip_address()
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.socket.bind((self.ip_address, self.port))
self.socket.listen(5)
if self.port == 0:
self.port = self.socket.getsockname()[1]
self.poller.add(self)
self.client_sockets = {}
self.listener.add_device(self)
"OPT: \"http://schemas.upnp.org/upnp/1/0/\"; ns=01\r\n"
"01-NLS: %s\r\n"
"SERVER: %s\r\n"
"ST: %s\r\n"
"USN: uuid:%s::%s\r\n" % (date_str, location_url, self.uuid, self.server_version, search_target, self.persistent_uuid, search_target))
if self.other_headers:
for header in self.other_headers:
message += "%s\r\n" % header
message += "\r\n"
temp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
temp_socket.sendto(message, destination)
# This subclass does the bulk of the work to mimic a WeMo switch on the network.
class fauxmo(upnp_device):
@staticmethod
def make_uuid(name):
return ''.join(["%x" % sum([ord(c) for c in name])] + ["%x" % ord(c) for c in "%sfauxmo!" % name])[:14]
def __init__(self, name, listener, poller, ip_address, port, action_handler = None):
self.serial = self.make_uuid(name)
self.name = name
self.ip_address = ip_address
persistent_uuid = "Socket-1_0-" + self.serial
other_headers = ['X-User-Agent: redsonic']
upnp_device.__init__(self, listener, poller, port, "http://%(ip_address)s:%(port)s/setup.xml", "Unspecified, UPnP/1.0, Unspecified", persistent_uuid, other_headers=other_headers, ip_address=ip_address)
if action_handler:
self.action_handler = action_handler
else:
self.action_handler = self
dbg("FauxMo device '%s' ready on %s:%s" % (self.name, self.ip_address, self.port))
def __init__(self, listener, poller, port, root_url, server_version, persistent_uuid, other_headers = None, ip_address = None):
self.listener = listener
self.poller = poller
self.port = port
self.root_url = root_url
self.server_version = server_version
self.persistent_uuid = persistent_uuid
self.uuid = uuid.uuid4()
self.other_headers = other_headers
if ip_address:
self.ip_address = ip_address
else:
self.ip_address = upnp_device.local_ip_address()
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.socket.bind((self.ip_address, self.port))
self.socket.listen(5)
if self.port == 0:
self.port = self.socket.getsockname()[1]
self.poller.add(self)
self.client_sockets = {}
self.listener.add_device(self)
def __init__(self, listener, poller, port, root_url, server_version, persistent_uuid, other_headers = None, ip_address = None, log = None):
self.listener = listener
self.poller = poller
self.port = port
self.root_url = root_url
self.server_version = server_version
self.persistent_uuid = persistent_uuid
self.uuid = uuid.uuid4()
self.other_headers = other_headers
if (log != None):
self.log = log
if ip_address:
self.ip_address = ip_address
else:
self.ip_address = upnp_device.local_ip_address()
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.socket.bind((self.ip_address, self.port))
self.socket.listen(5)
if self.port == 0:
self.port = self.socket.getsockname()[1]
self.poller.add(self)
self.client_sockets = {}
self.listener.add_device(self)