Cache emulated_hue local ip check (#102020)

homeassistant/components/emulated_hue/hue_api.py

@@ -121,6 +121,12 @@
 )
 
 
+@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 @@ def __init__(self, config: Config) -> None:
     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 @@ def __init__(self, config: Config) -> None:
     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 @@ def __init__(self, config: Config) -> None:
     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 @@ def __init__(self, config: Config) -> None:
     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 @@ def __init__(self, config: Config) -> None:
     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 @@ def __init__(self, config: Config) -> None:
     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 @@ async def put(  # noqa: C901
     ) -> 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

tests/components/emulated_hue/test_hue_api.py

@@ -36,9 +36,11 @@
     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."""