Backport core api filter (#4165)

requirements_tests.txt

@@ -14,3 +14,4 @@ pytest==7.2.1
 pyupgrade==3.3.1
 time-machine==2.9.0
 typing_extensions==4.3.0
+urllib3==1.26.14

supervisor/api/__init__.py

@@ -53,6 +53,7 @@ class RestAPI(CoreSysAttributes):
         self.webapp: web.Application = web.Application(
             client_max_size=MAX_CLIENT_SIZE,
             middlewares=[
+                self.security.block_bad_requests,
                 self.security.system_validation,
                 self.security.token_validation,
             ],

supervisor/api/middleware/security.py

@@ -1,9 +1,11 @@
 """Handle security part of this API."""
 import logging
 import re
+from typing import Final
+from urllib.parse import unquote
 
 from aiohttp.web import Request, RequestHandler, Response, middleware
-from aiohttp.web_exceptions import HTTPForbidden, HTTPUnauthorized
+from aiohttp.web_exceptions import HTTPBadRequest, HTTPForbidden, HTTPUnauthorized
 
 from ...const import (
     REQUEST_FROM,
@@ -22,7 +24,7 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
 # fmt: off
 
 # Block Anytime
-BLACKLIST = re.compile(
+BLACKLIST: Final = re.compile(
     r"^(?:"
     r"|/homeassistant/api/hassio/.*"
     r"|/core/api/hassio/.*"
@@ -30,7 +32,7 @@ BLACKLIST = re.compile(
 )
 
 # Free to call or have own security concepts
-NO_SECURITY_CHECK = re.compile(
+NO_SECURITY_CHECK: Final = re.compile(
     r"^(?:"
     r"|/homeassistant/api/.*"
     r"|/homeassistant/websocket"
@@ -41,14 +43,14 @@ NO_SECURITY_CHECK = re.compile(
 )
 
 # Observer allow API calls
-OBSERVER_CHECK = re.compile(
+OBSERVER_CHECK: Final = re.compile(
     r"^(?:"
     r"|/.+/info"
     r")$"
 )
 
 # Can called by every add-on
-ADDONS_API_BYPASS = re.compile(
+ADDONS_API_BYPASS: Final = re.compile(
     r"^(?:"
     r"|/addons/self/(?!security|update)[^/]+"
     r"|/addons/self/options/config"
@@ -60,7 +62,7 @@ ADDONS_API_BYPASS = re.compile(
 )
 
 # Policy role add-on API access
-ADDONS_ROLE_ACCESS = {
+ADDONS_ROLE_ACCESS: dict[str, re.Pattern] = {
     ROLE_DEFAULT: re.compile(
         r"^(?:"
         r"|/.+/info"
@@ -111,6 +113,26 @@ ADDONS_ROLE_ACCESS = {
     ),
 }
 
+FILTERS: Final = re.compile(
+    r"(?:"
+
+    # Common exploits
+    r"proc/self/environ"
+    r"|(<|%3C).*script.*(>|%3E)"
+
+    # File Injections
+    r"|(\.\.//?)+"  # ../../anywhere
+    r"|[a-zA-Z0-9_]=/([a-z0-9_.]//?)+"  # .html?v=/.//test
+
+    # SQL Injections
+    r"|union.*select.*\("
+    r"|union.*all.*select.*"
+    r"|concat.*\("
+
+    r")",
+    flags=re.IGNORECASE,
+)
+
 # fmt: on
 
 
@@ -121,6 +143,32 @@ class SecurityMiddleware(CoreSysAttributes):
         """Initialize security middleware."""
         self.coresys: CoreSys = coresys
 
+    def _recursive_unquote(self, value: str) -> str:
+        """Handle values that are encoded multiple times."""
+        if (unquoted := unquote(value)) != value:
+            unquoted = self._recursive_unquote(unquoted)
+        return unquoted
+
+    @middleware
+    async def block_bad_requests(
+        self, request: Request, handler: RequestHandler
+    ) -> Response:
+        """Process request and tblock commonly known exploit attempts."""
+        if FILTERS.search(self._recursive_unquote(request.path)):
+            _LOGGER.warning(
+                "Filtered a potential harmful request to: %s", request.raw_path
+            )
+            raise HTTPBadRequest
+
+        if FILTERS.search(self._recursive_unquote(request.query_string)):
+            _LOGGER.warning(
+                "Filtered a request with a potential harmful query string: %s",
+                request.raw_path,
+            )
+            raise HTTPBadRequest
+
+        return await handler(request)
+
     @middleware
     async def system_validation(
         self, request: Request, handler: RequestHandler

tests/api/middleware/test_security.py

@@ -1,8 +1,10 @@
 """Test API security layer."""
+from http import HTTPStatus
 from unittest.mock import patch
 
 from aiohttp import web
 import pytest
+import urllib3
 
 from supervisor.api import RestAPI
 from supervisor.const import CoreState
@@ -11,6 +13,11 @@ from supervisor.coresys import CoreSys
 # pylint: disable=redefined-outer-name
 
 
+async def mock_handler(request):
+    """Return OK."""
+    return web.Response(text="OK")
+
+
 @pytest.fixture
 async def api_system(aiohttp_client, run_dir, coresys: CoreSys):
     """Fixture for RestAPI client."""
@@ -20,7 +27,10 @@ async def api_system(aiohttp_client, run_dir, coresys: CoreSys):
         os.environ = {"SUPERVISOR_NAME": "hassio_supervisor"}
         await api.load()
 
+    api.webapp.middlewares.append(api.security.block_bad_requests)
     api.webapp.middlewares.append(api.security.system_validation)
+    api.webapp.router.add_get("/{all:.*}", mock_handler)
+
     yield await aiohttp_client(api.webapp)
 
 
@@ -62,3 +72,66 @@ async def test_api_security_system_startup(api_system, coresys: CoreSys):
 
     resp = await api_system.get("/supervisor/ping")
     assert resp.status == 200
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("request_path", "request_params", "fail_on_query_string"),
+    [
+        ("/proc/self/environ", {}, False),
+        ("/", {"test": "/test/../../api"}, True),
+        ("/", {"test": "test/../../api"}, True),
+        ("/", {"test": "/test/%2E%2E%2f%2E%2E%2fapi"}, True),
+        ("/", {"test": "test/%2E%2E%2f%2E%2E%2fapi"}, True),
+        ("/", {"test": "test/%252E%252E/api"}, True),
+        ("/", {"test": "test/%252E%252E%2fapi"}, True),
+        (
+            "/",
+            {"test": "test/%2525252E%2525252E%2525252f%2525252E%2525252E%2525252fapi"},
+            True,
+        ),
+        ("/test/.%252E/api", {}, False),
+        ("/test/%252E%252E/api", {}, False),
+        ("/test/%2E%2E%2f%2E%2E%2fapi", {}, False),
+        ("/test/%2525252E%2525252E%2525252f%2525252E%2525252E/api", {}, False),
+        ("/", {"sql": ";UNION SELECT (a, b"}, True),
+        ("/", {"sql": "UNION%20SELECT%20%28a%2C%20b"}, True),
+        ("/UNION%20SELECT%20%28a%2C%20b", {}, False),
+        ("/", {"sql": "concat(..."}, True),
+        ("/", {"xss": "<script >"}, True),
+        ("/<script >", {"xss": ""}, False),
+        ("/%3Cscript%3E", {}, False),
+    ],
+)
+async def test_bad_requests(
+    request_path,
+    request_params,
+    fail_on_query_string,
+    api_system,
+    caplog: pytest.LogCaptureFixture,
+    loop,
+) -> None:
+    """Test request paths that should be filtered."""
+
+    # Manual params handling
+    if request_params:
+        raw_params = "&".join(f"{val}={key}" for val, key in request_params.items())
+        man_params = f"?{raw_params}"
+    else:
+        man_params = ""
+
+    http = urllib3.PoolManager()
+    resp = await loop.run_in_executor(
+        None,
+        http.request,
+        "GET",
+        f"http://{api_system.host}:{api_system.port}{request_path}{man_params}",
+        request_params,
+    )
+
+    assert resp.status == HTTPStatus.BAD_REQUEST
+
+    message = "Filtered a potential harmful request to:"
+    if fail_on_query_string:
+        message = "Filtered a request with a potential harmful query string:"
+    assert message in caplog.text