Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"/eventservice.xml"
""
""
"urn:Belkin:service:metainfo:1"
"urn:Belkin:serviceId:metainfo1"
"/upnp/control/metainfo1"
"/upnp/event/metainfo1"
"/metainfoservice.xml"
""
""
""
""
)
setup_response = self.add_http_headers(setup_xml)
logger.debug(f"Fauxmo response to setup request:\n{setup_response}")
self.transport.write(setup_response.encode())
self.transport.close()
"""Run the main fauxmo process.
Spawns a UDP server to handle the Echo's UPnP / SSDP device discovery
process as well as multiple TCP servers to respond to the Echo's device
setup requests and handle its process for turning devices on and off.
Args:
config_path_str: Path to config file. If not given will search for
`config.json` in cwd, `~/.fauxmo/`, and
`/etc/fauxmo/`.
verbosity: Logging verbosity, defaults to 20
"""
logger.setLevel(verbosity)
logger.info(f"Fauxmo {__version__}")
logger.debug(sys.version)
if config_path_str:
config_path = pathlib.Path(config_path_str)
else:
for config_dir in (".", "~/.fauxmo", "/etc/fauxmo"):
config_path = pathlib.Path(config_dir).expanduser() / "config.json"
if config_path.is_file():
logger.info(f"Using config: {config_path}")
break
try:
config = json.loads(config_path.read_text())
except FileNotFoundError:
logger.error(
"Could not find config file in default search path. Try "
"specifying your file with `-c`.\n"
def datagram_received(
self, data: Union[bytes, Text], addr: Tuple[str, int]
) -> None:
"""Check incoming UDP data for requests for Wemo devices.
Args:
data: Incoming data content
addr: Address sending data
"""
if isinstance(data, bytes):
data = data.decode("utf8")
logger.debug(f"Received data below from {addr}:")
logger.debug(data)
discover_patterns = [
"ST: urn:Belkin:device:**",
"ST: upnp:rootdevice",
"ST: ssdp:all",
]
discover_pattern = next(
(pattern for pattern in discover_patterns if pattern in data), None
)
if (
'man: "ssdp:discover"' in data.lower()
and discover_pattern is not None
):
mx = 0.0
def datagram_received(
self, data: Union[bytes, Text], addr: Tuple[str, int]
) -> None:
"""Check incoming UDP data for requests for Wemo devices.
Args:
data: Incoming data content
addr: Address sending data
"""
if isinstance(data, bytes):
data = data.decode("utf8")
logger.debug(f"Received data below from {addr}:")
logger.debug(data)
discover_patterns = [
"ST: urn:Belkin:device:**",
"ST: upnp:rootdevice",
"ST: ssdp:all",
]
discover_pattern = next(
(pattern for pattern in discover_patterns if pattern in data), None
)
if (
'man: "ssdp:discover"' in data.lower()
and discover_pattern is not None
):
mx = 0.0
mx_line = next(
def get_local_ip(ip_address: str = None) -> str:
"""Attempt to get the local network-connected IP address.
Args:
ip_address: Either desired ip address or string or "auto"
Returns:
Current IP address as string
"""
if ip_address is None or ip_address.lower() == "auto":
logger.debug("Attempting to get IP address automatically")
hostname = socket.gethostname()
ip_address = socket.gethostbyname(hostname)
# Workaround for Linux returning localhost
# See: SO question #166506 by @UnkwnTech
if ip_address in ['127.0.1.1', '127.0.0.1', 'localhost']:
tempsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
tempsock.connect(('8.8.8.8', 0))
ip_address = tempsock.getsockname()[0]
tempsock.close()
logger.debug(f"Using IP address: {ip_address}")
return ip_address
try:
plugin = PluginClass(**plugin_vars, **device)
except TypeError:
logger.error(f"Error in plugin {repr(PluginClass)}")
raise
fauxmo = partial(Fauxmo, name=plugin.name, plugin=plugin)
coro = loop.create_server(fauxmo, host=fauxmo_ip, port=plugin.port)
server = loop.run_until_complete(coro)
server.fauxmoplugin = plugin # type: ignore
servers.append(server)
ssdp_server.add_device(plugin.name, fauxmo_ip, plugin.port)
logger.debug(f"Started fauxmo device: {repr(fauxmo.keywords)}")
logger.info("Starting UDP server")
listen = loop.create_datagram_endpoint(
lambda: ssdp_server, sock=make_udp_sock()
)
transport, _ = loop.run_until_complete(listen)
for signame in ("SIGINT", "SIGTERM"):
try:
loop.add_signal_handler(getattr(signal, signame), loop.stop)
# Workaround for Windows (https://github.com/n8henrie/fauxmo/issues/21)
except NotImplementedError:
if sys.platform == "win32":
pass
def connection_made(self, transport: asyncio.BaseTransport) -> None:
"""Accept an incoming TCP connection.
Args:
transport: Passed in asyncio.Transport
"""
peername = transport.get_extra_info("peername")
logger.debug(f"Connection made with: {peername}")
self.transport = cast(asyncio.Transport, transport)
"""
if ip_address is None or ip_address.lower() == "auto":
logger.debug("Attempting to get IP address automatically")
hostname = socket.gethostname()
ip_address = socket.gethostbyname(hostname)
# Workaround for Linux returning localhost
# See: SO question #166506 by @UnkwnTech
if ip_address in ['127.0.1.1', '127.0.0.1', 'localhost']:
tempsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
tempsock.connect(('8.8.8.8', 0))
ip_address = tempsock.getsockname()[0]
tempsock.close()
logger.debug(f"Using IP address: {ip_address}")
return ip_address