Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions homeassistant/components/emulated_hue/const.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
"""Constants for emulated_hue."""

HUE_SERIAL_NUMBER = "001788FFFE23BFC2"
HUE_UUID = "2f402f80-da50-11e1-9b23-001788255acc"
55 changes: 36 additions & 19 deletions homeassistant/components/emulated_hue/upnp.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
from homeassistant import core
from homeassistant.components.http import HomeAssistantView

from .const import HUE_SERIAL_NUMBER, HUE_UUID

_LOGGER = logging.getLogger(__name__)


Expand Down Expand Up @@ -42,8 +44,8 @@ def get(self, request):
<modelName>Philips hue bridge 2015</modelName>
<modelNumber>BSB002</modelNumber>
<modelURL>http://www.meethue.com</modelURL>
<serialNumber>001788FFFE23BFC2</serialNumber>
<UDN>uuid:2f402f80-da50-11e1-9b23-001788255acc</UDN>
<serialNumber>{HUE_SERIAL_NUMBER}</serialNumber>
<UDN>uuid:{HUE_UUID}</UDN>
</device>
</root>
"""
Expand All @@ -70,21 +72,8 @@ def __init__(
self.host_ip_addr = host_ip_addr
self.listen_port = listen_port
self.upnp_bind_multicast = upnp_bind_multicast

# Note that the double newline at the end of
# this string is required per the SSDP spec
resp_template = f"""HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://{advertise_ip}:{advertise_port}/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
hue-bridgeid: 001788FFFE23BFC2
ST: upnp:rootdevice
USN: uuid:2f402f80-da50-11e1-9b23-00178829d301::upnp:rootdevice

"""

self.upnp_response = resp_template.replace("\n", "\r\n").encode("utf-8")
self.advertise_ip = advertise_ip
self.advertise_port = advertise_port

def run(self):
"""Run the server."""
Expand Down Expand Up @@ -136,10 +125,13 @@ def run(self):
continue

if "M-SEARCH" in data.decode("utf-8", errors="ignore"):
_LOGGER.debug("UPNP Responder M-SEARCH method received: %s", data)
# SSDP M-SEARCH method received, respond to it with our info
resp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
response = self._handle_request(data)

resp_socket.sendto(self.upnp_response, addr)
resp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
resp_socket.sendto(response, addr)
_LOGGER.debug("UPNP Responder responding with: %s", response)
resp_socket.close()

def stop(self):
Expand All @@ -148,6 +140,31 @@ def stop(self):
self._interrupted = True
self.join()

def _handle_request(self, data):
if "upnp:rootdevice" in data.decode("utf-8", errors="ignore"):
return self._prepare_response(
"upnp:rootdevice", f"uuid:{HUE_UUID}::upnp:rootdevice"
)

return self._prepare_response(
"urn:schemas-upnp-org:device:basic:1", f"uuid:{HUE_UUID}"
)

def _prepare_response(self, search_target, unique_service_name):
# Note that the double newline at the end of
# this string is required per the SSDP spec
response = f"""HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://{self.advertise_ip}:{self.advertise_port}/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
hue-bridgeid: {HUE_SERIAL_NUMBER}
ST: {search_target}
USN: {unique_service_name}

"""
return response.replace("\n", "\r\n").encode("utf-8")


def clean_socket_close(sock):
"""Close a socket connection and logs its closure."""
Expand Down
60 changes: 60 additions & 0 deletions tests/components/emulated_hue/test_upnp.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,66 @@ def tearDownClass(cls):
"""Stop the class."""
cls.hass.stop()

def test_upnp_discovery_basic(self):
"""Tests the UPnP basic discovery response."""
with patch("threading.Thread.__init__"):
upnp_responder_thread = emulated_hue.UPNPResponderThread(
"0.0.0.0", 80, True, "192.0.2.42", 8080
)

"""Original request emitted by the Hue Bridge v1 app."""
request = """M-SEARCH * HTTP/1.1
HOST:239.255.255.250:1900
ST:ssdp:all
Man:"ssdp:discover"
MX:3

"""
encoded_request = request.replace("\n", "\r\n").encode("utf-8")

response = upnp_responder_thread._handle_request(encoded_request)
expected_response = """HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://192.0.2.42:8080/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
hue-bridgeid: 001788FFFE23BFC2
ST: urn:schemas-upnp-org:device:basic:1
USN: uuid:2f402f80-da50-11e1-9b23-001788255acc

"""
assert expected_response.replace("\n", "\r\n").encode("utf-8") == response

def test_upnp_discovery_rootdevice(self):
"""Tests the UPnP rootdevice discovery response."""
with patch("threading.Thread.__init__"):
upnp_responder_thread = emulated_hue.UPNPResponderThread(
"0.0.0.0", 80, True, "192.0.2.42", 8080
)

"""Original request emitted by Busch-Jaeger free@home SysAP."""
request = """M-SEARCH * HTTP/1.1
HOST: 239.255.255.250:1900
MAN: "ssdp:discover"
MX: 40
ST: upnp:rootdevice

"""
encoded_request = request.replace("\n", "\r\n").encode("utf-8")

response = upnp_responder_thread._handle_request(encoded_request)
expected_response = """HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://192.0.2.42:8080/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
hue-bridgeid: 001788FFFE23BFC2
ST: upnp:rootdevice
USN: uuid:2f402f80-da50-11e1-9b23-001788255acc::upnp:rootdevice

"""
assert expected_response.replace("\n", "\r\n").encode("utf-8") == response

def test_description_xml(self):
"""Test the description."""
result = requests.get(BRIDGE_URL_BASE.format("/description.xml"), timeout=5)
Expand Down