From 547c38a515c98db72b59c3b45191ceda3256b5ed Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sat, 14 Oct 2023 16:19:10 -1000 Subject: [PATCH] Cache emulated_hue local ip check (#102020) --- .../components/emulated_hue/hue_api.py | 22 ++++++++++++------- tests/components/emulated_hue/test_hue_api.py | 15 ++++++++++++- 2 files changed, 28 insertions(+), 9 deletions(-) diff --git a/homeassistant/components/emulated_hue/hue_api.py b/homeassistant/components/emulated_hue/hue_api.py index 566779671e8..6dfd49c371c 100644 --- a/homeassistant/components/emulated_hue/hue_api.py +++ b/homeassistant/components/emulated_hue/hue_api.py @@ -121,6 +121,12 @@ DIMMABLE_SUPPORT_FEATURES = ( ) +@lru_cache(maxsize=32) +def _remote_is_allowed(address: str) -> bool: + """Check if remote address is allowed.""" + return is_local(ip_address(address)) + + class HueUnauthorizedUser(HomeAssistantView): """Handle requests to find the emulated hue bridge.""" @@ -145,7 +151,7 @@ class HueUsernameView(HomeAssistantView): async def post(self, request: web.Request) -> web.Response: """Handle a POST request.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) try: @@ -174,7 +180,7 @@ class HueAllGroupsStateView(HomeAssistantView): def get(self, request: web.Request, username: str) -> web.Response: """Process a request to make the Brilliant Lightpad work.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) return self.json({}) @@ -195,7 +201,7 @@ class HueGroupView(HomeAssistantView): def put(self, request: web.Request, username: str) -> web.Response: """Process a request to make the Logitech Pop working.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) return self.json( @@ -226,7 +232,7 @@ class HueAllLightsStateView(HomeAssistantView): def get(self, request: web.Request, username: str) -> web.Response: """Process a request to get the list of available lights.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) return self.json(create_list_of_entities(self.config, request)) @@ -247,7 +253,7 @@ class HueFullStateView(HomeAssistantView): def get(self, request: web.Request, username: str) -> web.Response: """Process a request to get the list of available lights.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("only local IPs allowed", HTTPStatus.UNAUTHORIZED) if username != HUE_API_USERNAME: return self.json(UNAUTHORIZED_USER) @@ -276,7 +282,7 @@ class HueConfigView(HomeAssistantView): def get(self, request: web.Request, username: str = "") -> web.Response: """Process a request to get the configuration.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("only local IPs allowed", HTTPStatus.UNAUTHORIZED) json_response = create_config_model(self.config, request) @@ -299,7 +305,7 @@ class HueOneLightStateView(HomeAssistantView): def get(self, request: web.Request, username: str, entity_id: str) -> web.Response: """Process a request to get the state of an individual light.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) hass: core.HomeAssistant = request.app["hass"] @@ -341,7 +347,7 @@ class HueOneLightChangeView(HomeAssistantView): ) -> web.Response: """Process a request to set the state of an individual light.""" assert request.remote is not None - if not is_local(ip_address(request.remote)): + if not _remote_is_allowed(request.remote): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) config = self.config diff --git a/tests/components/emulated_hue/test_hue_api.py b/tests/components/emulated_hue/test_hue_api.py index 24acde0709a..fb5ff265497 100644 --- a/tests/components/emulated_hue/test_hue_api.py +++ b/tests/components/emulated_hue/test_hue_api.py @@ -36,9 +36,11 @@ from homeassistant.components.emulated_hue.hue_api import ( HueAllLightsStateView, HueConfigView, HueFullStateView, + HueGroupView, HueOneLightChangeView, HueOneLightStateView, HueUsernameView, + _remote_is_allowed, ) from homeassistant.const import ( ATTR_ENTITY_ID, @@ -232,6 +234,7 @@ def _mock_hue_endpoints( HueOneLightStateView(config).register(hass, web_app, web_app.router) HueOneLightChangeView(config).register(hass, web_app, web_app.router) HueAllGroupsStateView(config).register(hass, web_app, web_app.router) + HueGroupView(config).register(hass, web_app, web_app.router) HueFullStateView(config).register(hass, web_app, web_app.router) HueConfigView(config).register(hass, web_app, web_app.router) @@ -1327,23 +1330,33 @@ async def test_external_ip_blocked(hue_client) -> None: "/api/username/lights/light.ceiling_lights", ] postUrls = ["/api"] - putUrls = ["/api/username/lights/light.ceiling_lights/state"] + putUrls = [ + "/api/username/lights/light.ceiling_lights/state", + "/api/username/groups/0/action", + ] with patch( "homeassistant.components.emulated_hue.hue_api.ip_address", return_value=ip_address("45.45.45.45"), ): for getUrl in getUrls: + _remote_is_allowed.cache_clear() result = await hue_client.get(getUrl) assert result.status == HTTPStatus.UNAUTHORIZED for postUrl in postUrls: + _remote_is_allowed.cache_clear() result = await hue_client.post(postUrl) assert result.status == HTTPStatus.UNAUTHORIZED for putUrl in putUrls: + _remote_is_allowed.cache_clear() result = await hue_client.put(putUrl) assert result.status == HTTPStatus.UNAUTHORIZED + # We are patching inside of a cache so be sure to clear it + # so that the next test is not affected + _remote_is_allowed.cache_clear() + async def test_unauthorized_user_blocked(hue_client) -> None: """Test unauthorized_user blocked."""