diff --git a/homeassistant/components/hassio/const.py b/homeassistant/components/hassio/const.py
index 64ef7a718a5..2710e146540 100644
--- a/homeassistant/components/hassio/const.py
+++ b/homeassistant/components/hassio/const.py
@@ -36,6 +36,7 @@ X_AUTH_TOKEN = "X-Supervisor-Token"
 X_INGRESS_PATH = "X-Ingress-Path"
 X_HASS_USER_ID = "X-Hass-User-ID"
 X_HASS_IS_ADMIN = "X-Hass-Is-Admin"
+X_HASS_SOURCE = "X-Hass-Source"
 
 WS_TYPE = "type"
 WS_ID = "id"
diff --git a/homeassistant/components/hassio/handler.py b/homeassistant/components/hassio/handler.py
index 0d923075bf7..762df4f79ca 100644
--- a/homeassistant/components/hassio/handler.py
+++ b/homeassistant/components/hassio/handler.py
@@ -17,7 +17,7 @@ from homeassistant.const import SERVER_PORT
 from homeassistant.core import HomeAssistant
 from homeassistant.loader import bind_hass
 
-from .const import ATTR_DISCOVERY, DOMAIN
+from .const import ATTR_DISCOVERY, DOMAIN, X_HASS_SOURCE
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -445,6 +445,8 @@ class HassIO:
         payload=None,
         timeout=10,
         return_text=False,
+        *,
+        source="core.handler",
     ):
         """Send API command to Hass.io.
 
@@ -458,7 +460,8 @@ class HassIO:
                 headers={
                     aiohttp.hdrs.AUTHORIZATION: (
                         f"Bearer {os.environ.get('SUPERVISOR_TOKEN', '')}"
-                    )
+                    ),
+                    X_HASS_SOURCE: source,
                 },
                 timeout=aiohttp.ClientTimeout(total=timeout),
             )
diff --git a/homeassistant/components/hassio/http.py b/homeassistant/components/hassio/http.py
index 2b7145bdcaa..8a8583a7daf 100644
--- a/homeassistant/components/hassio/http.py
+++ b/homeassistant/components/hassio/http.py
@@ -6,6 +6,7 @@ from http import HTTPStatus
 import logging
 import os
 import re
+from urllib.parse import quote, unquote
 
 import aiohttp
 from aiohttp import web
@@ -19,13 +20,16 @@ from aiohttp.hdrs import (
     TRANSFER_ENCODING,
 )
 from aiohttp.web_exceptions import HTTPBadGateway
-from multidict import istr
 
-from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
+from homeassistant.components.http import (
+    KEY_AUTHENTICATED,
+    KEY_HASS_USER,
+    HomeAssistantView,
+)
 from homeassistant.components.onboarding import async_is_onboarded
 from homeassistant.core import HomeAssistant
 
-from .const import X_HASS_IS_ADMIN, X_HASS_USER_ID
+from .const import X_HASS_SOURCE
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -34,23 +38,53 @@ MAX_UPLOAD_SIZE = 1024 * 1024 * 1024
 # pylint: disable=implicit-str-concat
 NO_TIMEOUT = re.compile(
     r"^(?:"
-    r"|homeassistant/update"
-    r"|hassos/update"
-    r"|hassos/update/cli"
-    r"|supervisor/update"
-    r"|addons/[^/]+/(?:update|install|rebuild)"
     r"|backups/.+/full"
     r"|backups/.+/partial"
     r"|backups/[^/]+/(?:upload|download)"
     r")$"
 )
 
-NO_AUTH_ONBOARDING = re.compile(r"^(?:" r"|supervisor/logs" r"|backups/[^/]+/.+" r")$")
+# fmt: off
+# Onboarding can upload backups and restore it
+PATHS_NOT_ONBOARDED = re.compile(
+    r"^(?:"
+    r"|backups/[a-f0-9]{8}(/info|/new/upload|/download|/restore/full|/restore/partial)?"
+    r"|backups/new/upload"
+    r")$"
+)
 
-NO_AUTH = re.compile(r"^(?:" r"|app/.*" r"|[store\/]*addons/[^/]+/(logo|icon)" r")$")
+# Authenticated users manage backups + download logs
+PATHS_ADMIN = re.compile(
+    r"^(?:"
+    r"|backups/[a-f0-9]{8}(/info|/download|/restore/full|/restore/partial)?"
+    r"|backups/new/upload"
+    r"|audio/logs"
+    r"|cli/logs"
+    r"|core/logs"
+    r"|dns/logs"
+    r"|host/logs"
+    r"|multicast/logs"
+    r"|observer/logs"
+    r"|supervisor/logs"
+    r"|addons/[^/]+/logs"
+    r")$"
+)
 
-NO_STORE = re.compile(r"^(?:" r"|app/entrypoint.js" r")$")
+# Unauthenticated requests come in for Supervisor panel + add-on images
+PATHS_NO_AUTH = re.compile(
+    r"^(?:"
+    r"|app/.*"
+    r"|(store/)?addons/[^/]+/(logo|icon)"
+    r")$"
+)
+
+NO_STORE = re.compile(
+    r"^(?:"
+    r"|app/entrypoint.js"
+    r")$"
+)
 # pylint: enable=implicit-str-concat
+# fmt: on
 
 
 class HassIOView(HomeAssistantView):
@@ -65,38 +99,66 @@ class HassIOView(HomeAssistantView):
         self._host = host
         self._websession = websession
 
-    async def _handle(
-        self, request: web.Request, path: str
-    ) -> web.Response | web.StreamResponse:
-        """Route data to Hass.io."""
-        hass = request.app["hass"]
-        if _need_auth(hass, path) and not request[KEY_AUTHENTICATED]:
-            return web.Response(status=HTTPStatus.UNAUTHORIZED)
-
-        return await self._command_proxy(path, request)
-
-    delete = _handle
-    get = _handle
-    post = _handle
-
-    async def _command_proxy(
-        self, path: str, request: web.Request
-    ) -> web.StreamResponse:
+    async def _handle(self, request: web.Request, path: str) -> web.StreamResponse:
         """Return a client request with proxy origin for Hass.io supervisor.
 
-        This method is a coroutine.
+        Use cases:
+        - Onboarding allows restoring backups
+        - Load Supervisor panel and add-on logo unauthenticated
+        - User upload/restore backups
         """
-        headers = _init_header(request)
-        if path == "backups/new/upload":
-            # We need to reuse the full content type that includes the boundary
-            headers[
-                CONTENT_TYPE
-            ] = request._stored_content_type  # pylint: disable=protected-access
+        # No bullshit
+        if path != unquote(path):
+            return web.Response(status=HTTPStatus.BAD_REQUEST)
+
+        hass: HomeAssistant = request.app["hass"]
+        is_admin = request[KEY_AUTHENTICATED] and request[KEY_HASS_USER].is_admin
+        authorized = is_admin
+
+        if is_admin:
+            allowed_paths = PATHS_ADMIN
+
+        elif not async_is_onboarded(hass):
+            allowed_paths = PATHS_NOT_ONBOARDED
+
+            # During onboarding we need the user to manage backups
+            authorized = True
+
+        else:
+            # Either unauthenticated or not an admin
+            allowed_paths = PATHS_NO_AUTH
+
+        no_auth_path = PATHS_NO_AUTH.match(path)
+        headers = {
+            X_HASS_SOURCE: "core.http",
+        }
+
+        if no_auth_path:
+            if request.method != "GET":
+                return web.Response(status=HTTPStatus.METHOD_NOT_ALLOWED)
+
+        else:
+            if not allowed_paths.match(path):
+                return web.Response(status=HTTPStatus.UNAUTHORIZED)
+
+            if authorized:
+                headers[
+                    AUTHORIZATION
+                ] = f"Bearer {os.environ.get('SUPERVISOR_TOKEN', '')}"
+
+            if request.method == "POST":
+                headers[CONTENT_TYPE] = request.content_type
+                # _stored_content_type is only computed once `content_type` is accessed
+                if path == "backups/new/upload":
+                    # We need to reuse the full content type that includes the boundary
+                    headers[
+                        CONTENT_TYPE
+                    ] = request._stored_content_type  # pylint: disable=protected-access
 
         try:
             client = await self._websession.request(
                 method=request.method,
-                url=f"http://{self._host}/{path}",
+                url=f"http://{self._host}/{quote(path)}",
                 params=request.query,
                 data=request.content,
                 headers=headers,
@@ -123,20 +185,8 @@ class HassIOView(HomeAssistantView):
 
         raise HTTPBadGateway()
 
-
-def _init_header(request: web.Request) -> dict[istr, str]:
-    """Create initial header."""
-    headers = {
-        AUTHORIZATION: f"Bearer {os.environ.get('SUPERVISOR_TOKEN', '')}",
-        CONTENT_TYPE: request.content_type,
-    }
-
-    # Add user data
-    if request.get("hass_user") is not None:
-        headers[istr(X_HASS_USER_ID)] = request["hass_user"].id
-        headers[istr(X_HASS_IS_ADMIN)] = str(int(request["hass_user"].is_admin))
-
-    return headers
+    get = _handle
+    post = _handle
 
 
 def _response_header(response: aiohttp.ClientResponse, path: str) -> dict[str, str]:
@@ -164,12 +214,3 @@ def _get_timeout(path: str) -> ClientTimeout:
     if NO_TIMEOUT.match(path):
         return ClientTimeout(connect=10, total=None)
     return ClientTimeout(connect=10, total=300)
-
-
-def _need_auth(hass: HomeAssistant, path: str) -> bool:
-    """Return if a path need authentication."""
-    if not async_is_onboarded(hass) and NO_AUTH_ONBOARDING.match(path):
-        return False
-    if NO_AUTH.match(path):
-        return False
-    return True
diff --git a/homeassistant/components/hassio/ingress.py b/homeassistant/components/hassio/ingress.py
index dceff75bca8..334c7cf719c 100644
--- a/homeassistant/components/hassio/ingress.py
+++ b/homeassistant/components/hassio/ingress.py
@@ -3,20 +3,22 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Iterable
+from functools import lru_cache
 from ipaddress import ip_address
 import logging
-import os
+from urllib.parse import quote
 
 import aiohttp
 from aiohttp import ClientTimeout, hdrs, web
 from aiohttp.web_exceptions import HTTPBadGateway, HTTPBadRequest
 from multidict import CIMultiDict
+from yarl import URL
 
 from homeassistant.components.http import HomeAssistantView
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
-from .const import X_AUTH_TOKEN, X_INGRESS_PATH
+from .const import X_HASS_SOURCE, X_INGRESS_PATH
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -42,9 +44,19 @@ class HassIOIngress(HomeAssistantView):
         self._host = host
         self._websession = websession
 
+    @lru_cache
     def _create_url(self, token: str, path: str) -> str:
         """Create URL to service."""
-        return f"http://{self._host}/ingress/{token}/{path}"
+        base_path = f"/ingress/{token}/"
+        url = f"http://{self._host}{base_path}{quote(path)}"
+
+        try:
+            if not URL(url).path.startswith(base_path):
+                raise HTTPBadRequest()
+        except ValueError as err:
+            raise HTTPBadRequest() from err
+
+        return url
 
     async def _handle(
         self, request: web.Request, token: str, path: str
@@ -185,10 +197,8 @@ def _init_header(request: web.Request, token: str) -> CIMultiDict | dict[str, st
             continue
         headers[name] = value
 
-    # Inject token / cleanup later on Supervisor
-    headers[X_AUTH_TOKEN] = os.environ.get("SUPERVISOR_TOKEN", "")
-
     # Ingress information
+    headers[X_HASS_SOURCE] = "core.ingress"
     headers[X_INGRESS_PATH] = f"/api/hassio_ingress/{token}"
 
     # Set X-Forwarded-For
diff --git a/homeassistant/components/hassio/websocket_api.py b/homeassistant/components/hassio/websocket_api.py
index 3670d5ca1fd..8a9a145f2d6 100644
--- a/homeassistant/components/hassio/websocket_api.py
+++ b/homeassistant/components/hassio/websocket_api.py
@@ -116,6 +116,7 @@ async def websocket_supervisor_api(
             method=msg[ATTR_METHOD],
             timeout=msg.get(ATTR_TIMEOUT, 10),
             payload=msg.get(ATTR_DATA, {}),
+            source="core.websocket_api",
         )
 
         if result.get(ATTR_RESULT) == "error":
diff --git a/tests/components/hassio/conftest.py b/tests/components/hassio/conftest.py
index a6cd956c95e..78ae9643d68 100644
--- a/tests/components/hassio/conftest.py
+++ b/tests/components/hassio/conftest.py
@@ -1,5 +1,6 @@
 """Fixtures for Hass.io."""
 import os
+import re
 from unittest.mock import Mock, patch
 
 import pytest
@@ -12,6 +13,16 @@ from homeassistant.setup import async_setup_component
 from . import SUPERVISOR_TOKEN
 
 
+@pytest.fixture(autouse=True)
+def disable_security_filter():
+    """Disable the security filter to ensure the integration is secure."""
+    with patch(
+        "homeassistant.components.http.security_filter.FILTERS",
+        re.compile("not-matching-anything"),
+    ):
+        yield
+
+
 @pytest.fixture
 def hassio_env():
     """Fixture to inject hassio env."""
@@ -37,6 +48,13 @@ def hassio_stubs(hassio_env, hass, hass_client, aioclient_mock):
     ), patch(
         "homeassistant.components.hassio.HassIO.get_info",
         side_effect=HassioAPIError(),
+    ), patch(
+        "homeassistant.components.hassio.HassIO.get_ingress_panels",
+        return_value={"panels": []},
+    ), patch(
+        "homeassistant.components.hassio.repairs.SupervisorRepairs.setup"
+    ), patch(
+        "homeassistant.components.hassio.HassIO.refresh_updates"
     ):
         hass.state = CoreState.starting
         hass.loop.run_until_complete(async_setup_component(hass, "hassio", {}))
@@ -67,13 +85,7 @@ async def hassio_client_supervisor(hass, aiohttp_client, hassio_stubs):
 
 
 @pytest.fixture
-def hassio_handler(hass, aioclient_mock):
+async def hassio_handler(hass, aioclient_mock):
     """Create mock hassio handler."""
-
-    async def get_client_session():
-        return async_get_clientsession(hass)
-
-    websession = hass.loop.run_until_complete(get_client_session())
-
     with patch.dict(os.environ, {"SUPERVISOR_TOKEN": SUPERVISOR_TOKEN}):
-        yield HassIO(hass.loop, websession, "127.0.0.1")
+        yield HassIO(hass.loop, async_get_clientsession(hass), "127.0.0.1")
diff --git a/tests/components/hassio/test_handler.py b/tests/components/hassio/test_handler.py
index ee23d5d350e..64e9e1c31cc 100644
--- a/tests/components/hassio/test_handler.py
+++ b/tests/components/hassio/test_handler.py
@@ -1,13 +1,21 @@
 """The tests for the hassio component."""
+from __future__ import annotations
+
+from typing import Any, Literal
+
 import aiohttp
+from aiohttp import hdrs, web
 import pytest
 
-from homeassistant.components.hassio.handler import HassioAPIError
+from homeassistant.components.hassio.handler import HassIO, HassioAPIError
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
 from tests.test_util.aiohttp import AiohttpClientMocker
 
 
-async def test_api_ping(hassio_handler, aioclient_mock: AiohttpClientMocker) -> None:
+async def test_api_ping(
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
+) -> None:
     """Test setup with API ping."""
     aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={"result": "ok"})
 
@@ -16,7 +24,7 @@ async def test_api_ping(hassio_handler, aioclient_mock: AiohttpClientMocker) ->
 
 
 async def test_api_ping_error(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API ping error."""
     aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={"result": "error"})
@@ -26,7 +34,7 @@ async def test_api_ping_error(
 
 
 async def test_api_ping_exeption(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API ping exception."""
     aioclient_mock.get("http://127.0.0.1/supervisor/ping", exc=aiohttp.ClientError())
@@ -35,7 +43,9 @@ async def test_api_ping_exeption(
     assert aioclient_mock.call_count == 1
 
 
-async def test_api_info(hassio_handler, aioclient_mock: AiohttpClientMocker) -> None:
+async def test_api_info(
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
+) -> None:
     """Test setup with API generic info."""
     aioclient_mock.get(
         "http://127.0.0.1/info",
@@ -53,7 +63,7 @@ async def test_api_info(hassio_handler, aioclient_mock: AiohttpClientMocker) ->
 
 
 async def test_api_info_error(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Home Assistant info error."""
     aioclient_mock.get(
@@ -67,7 +77,7 @@ async def test_api_info_error(
 
 
 async def test_api_host_info(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Host info."""
     aioclient_mock.get(
@@ -90,7 +100,7 @@ async def test_api_host_info(
 
 
 async def test_api_supervisor_info(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Supervisor info."""
     aioclient_mock.get(
@@ -108,7 +118,9 @@ async def test_api_supervisor_info(
     assert data["channel"] == "stable"
 
 
-async def test_api_os_info(hassio_handler, aioclient_mock: AiohttpClientMocker) -> None:
+async def test_api_os_info(
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
+) -> None:
     """Test setup with API OS info."""
     aioclient_mock.get(
         "http://127.0.0.1/os/info",
@@ -125,7 +137,7 @@ async def test_api_os_info(hassio_handler, aioclient_mock: AiohttpClientMocker)
 
 
 async def test_api_host_info_error(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Home Assistant info error."""
     aioclient_mock.get(
@@ -139,7 +151,7 @@ async def test_api_host_info_error(
 
 
 async def test_api_core_info(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Home Assistant Core info."""
     aioclient_mock.get(
@@ -153,7 +165,7 @@ async def test_api_core_info(
 
 
 async def test_api_core_info_error(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Home Assistant Core info error."""
     aioclient_mock.get(
@@ -167,7 +179,7 @@ async def test_api_core_info_error(
 
 
 async def test_api_homeassistant_stop(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Home Assistant stop."""
     aioclient_mock.post("http://127.0.0.1/homeassistant/stop", json={"result": "ok"})
@@ -177,7 +189,7 @@ async def test_api_homeassistant_stop(
 
 
 async def test_api_homeassistant_restart(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Home Assistant restart."""
     aioclient_mock.post("http://127.0.0.1/homeassistant/restart", json={"result": "ok"})
@@ -187,7 +199,7 @@ async def test_api_homeassistant_restart(
 
 
 async def test_api_addon_info(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Add-on info."""
     aioclient_mock.get(
@@ -201,7 +213,7 @@ async def test_api_addon_info(
 
 
 async def test_api_addon_stats(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Add-on stats."""
     aioclient_mock.get(
@@ -215,7 +227,7 @@ async def test_api_addon_stats(
 
 
 async def test_api_discovery_message(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API discovery message."""
     aioclient_mock.get(
@@ -229,7 +241,7 @@ async def test_api_discovery_message(
 
 
 async def test_api_retrieve_discovery(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API discovery message."""
     aioclient_mock.get(
@@ -243,7 +255,7 @@ async def test_api_retrieve_discovery(
 
 
 async def test_api_ingress_panels(
-    hassio_handler, aioclient_mock: AiohttpClientMocker
+    hassio_handler: HassIO, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test setup with API Ingress panels."""
     aioclient_mock.get(
@@ -267,3 +279,56 @@ async def test_api_ingress_panels(
     assert aioclient_mock.call_count == 1
     assert data["panels"]
     assert "slug" in data["panels"]
+
+
+@pytest.mark.parametrize(
+    ("api_call", "method", "payload"),
+    [
+        ["retrieve_discovery_messages", "GET", None],
+        ["refresh_updates", "POST", None],
+        ["update_diagnostics", "POST", True],
+    ],
+)
+async def test_api_headers(
+    hass,
+    aiohttp_raw_server,
+    socket_enabled,
+    api_call: str,
+    method: Literal["GET", "POST"],
+    payload: Any,
+) -> None:
+    """Test headers are forwarded correctly."""
+    received_request = None
+
+    async def mock_handler(request):
+        """Return OK."""
+        nonlocal received_request
+        received_request = request
+        return web.json_response({"result": "ok", "data": None})
+
+    server = await aiohttp_raw_server(mock_handler)
+    hassio_handler = HassIO(
+        hass.loop,
+        async_get_clientsession(hass),
+        f"{server.host}:{server.port}",
+    )
+
+    api_func = getattr(hassio_handler, api_call)
+    if payload:
+        await api_func(payload)
+    else:
+        await api_func()
+    assert received_request is not None
+
+    assert received_request.method == method
+    assert received_request.headers.get("X-Hass-Source") == "core.handler"
+
+    if method == "GET":
+        assert hdrs.CONTENT_TYPE not in received_request.headers
+        return
+
+    assert hdrs.CONTENT_TYPE in received_request.headers
+    if payload:
+        assert received_request.headers[hdrs.CONTENT_TYPE] == "application/json"
+    else:
+        assert received_request.headers[hdrs.CONTENT_TYPE] == "application/octet-stream"
diff --git a/tests/components/hassio/test_http.py b/tests/components/hassio/test_http.py
index 8ef6fa4001b..cb1dd639ec6 100644
--- a/tests/components/hassio/test_http.py
+++ b/tests/components/hassio/test_http.py
@@ -1,63 +1,45 @@
 """The tests for the hassio component."""
 import asyncio
 from http import HTTPStatus
+from unittest.mock import patch
 
 from aiohttp import StreamReader
 import pytest
 
-from homeassistant.components.hassio.http import _need_auth
-from homeassistant.core import HomeAssistant
-
-from tests.common import MockUser
 from tests.test_util.aiohttp import AiohttpClientMocker
 
 
-async def test_forward_request(
-    hassio_client, aioclient_mock: AiohttpClientMocker
-) -> None:
-    """Test fetching normal path."""
-    aioclient_mock.post("http://127.0.0.1/beer", text="response")
+@pytest.fixture
+def mock_not_onboarded():
+    """Mock that we're not onboarded."""
+    with patch(
+        "homeassistant.components.hassio.http.async_is_onboarded", return_value=False
+    ):
+        yield
 
-    resp = await hassio_client.post("/api/hassio/beer")
 
-    # Check we got right response
-    assert resp.status == HTTPStatus.OK
-    body = await resp.text()
-    assert body == "response"
-
-    # Check we forwarded command
-    assert len(aioclient_mock.mock_calls) == 1
+@pytest.fixture
+def hassio_user_client(hassio_client, hass_admin_user):
+    """Return a Hass.io HTTP client tied to a non-admin user."""
+    hass_admin_user.groups = []
+    return hassio_client
 
 
 @pytest.mark.parametrize(
-    "build_type", ["supervisor/info", "homeassistant/update", "host/info"]
-)
-async def test_auth_required_forward_request(hassio_noauth_client, build_type) -> None:
-    """Test auth required for normal request."""
-    resp = await hassio_noauth_client.post(f"/api/hassio/{build_type}")
-
-    # Check we got right response
-    assert resp.status == HTTPStatus.UNAUTHORIZED
-
-
-@pytest.mark.parametrize(
-    "build_type",
+    "path",
     [
-        "app/index.html",
-        "app/hassio-app.html",
-        "app/index.html",
-        "app/hassio-app.html",
-        "app/some-chunk.js",
-        "app/app.js",
+        "app/entrypoint.js",
+        "addons/bl_b392/logo",
+        "addons/bl_b392/icon",
     ],
 )
-async def test_forward_request_no_auth_for_panel(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+async def test_forward_request_onboarded_user_get(
+    hassio_user_client, aioclient_mock: AiohttpClientMocker, path: str
 ) -> None:
-    """Test no auth needed for ."""
-    aioclient_mock.get(f"http://127.0.0.1/{build_type}", text="response")
+    """Test fetching normal path."""
+    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")
 
-    resp = await hassio_client.get(f"/api/hassio/{build_type}")
+    resp = await hassio_user_client.get(f"/api/hassio/{path}")
 
     # Check we got right response
     assert resp.status == HTTPStatus.OK
@@ -66,15 +48,68 @@ async def test_forward_request_no_auth_for_panel(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
+    # We only expect a single header.
+    assert aioclient_mock.mock_calls[0][3] == {"X-Hass-Source": "core.http"}
 
 
-async def test_forward_request_no_auth_for_logo(
-    hassio_client, aioclient_mock: AiohttpClientMocker
+@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE", "RANDOM"])
+async def test_forward_request_onboarded_user_unallowed_methods(
+    hassio_user_client, aioclient_mock: AiohttpClientMocker, method: str
 ) -> None:
-    """Test no auth needed for logo."""
-    aioclient_mock.get("http://127.0.0.1/addons/bl_b392/logo", text="response")
+    """Test fetching normal path."""
+    resp = await hassio_user_client.post("/api/hassio/app/entrypoint.js")
 
-    resp = await hassio_client.get("/api/hassio/addons/bl_b392/logo")
+    # Check we got right response
+    assert resp.status == HTTPStatus.METHOD_NOT_ALLOWED
+
+    # Check we did not forward command
+    assert len(aioclient_mock.mock_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("bad_path", "expected_status"),
+    [
+        # Caught by bullshit filter
+        ("app/%252E./entrypoint.js", HTTPStatus.BAD_REQUEST),
+        # The .. is processed, making it an unauthenticated path
+        ("app/../entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        ("app/%2E%2E/entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        # Unauthenticated path
+        ("supervisor/info", HTTPStatus.UNAUTHORIZED),
+        ("supervisor/logs", HTTPStatus.UNAUTHORIZED),
+        ("addons/bl_b392/logs", HTTPStatus.UNAUTHORIZED),
+    ],
+)
+async def test_forward_request_onboarded_user_unallowed_paths(
+    hassio_user_client,
+    aioclient_mock: AiohttpClientMocker,
+    bad_path: str,
+    expected_status: int,
+) -> None:
+    """Test fetching normal path."""
+    resp = await hassio_user_client.get(f"/api/hassio/{bad_path}")
+
+    # Check we got right response
+    assert resp.status == expected_status
+    # Check we didn't forward command
+    assert len(aioclient_mock.mock_calls) == 0
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "app/entrypoint.js",
+        "addons/bl_b392/logo",
+        "addons/bl_b392/icon",
+    ],
+)
+async def test_forward_request_onboarded_noauth_get(
+    hassio_noauth_client, aioclient_mock: AiohttpClientMocker, path: str
+) -> None:
+    """Test fetching normal path."""
+    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")
+
+    resp = await hassio_noauth_client.get(f"/api/hassio/{path}")
 
     # Check we got right response
     assert resp.status == HTTPStatus.OK
@@ -83,15 +118,73 @@ async def test_forward_request_no_auth_for_logo(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
+    # We only expect a single header.
+    assert aioclient_mock.mock_calls[0][3] == {"X-Hass-Source": "core.http"}
 
 
-async def test_forward_request_no_auth_for_icon(
-    hassio_client, aioclient_mock: AiohttpClientMocker
+@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE", "RANDOM"])
+async def test_forward_request_onboarded_noauth_unallowed_methods(
+    hassio_noauth_client, aioclient_mock: AiohttpClientMocker, method: str
 ) -> None:
-    """Test no auth needed for icon."""
-    aioclient_mock.get("http://127.0.0.1/addons/bl_b392/icon", text="response")
+    """Test fetching normal path."""
+    resp = await hassio_noauth_client.post("/api/hassio/app/entrypoint.js")
 
-    resp = await hassio_client.get("/api/hassio/addons/bl_b392/icon")
+    # Check we got right response
+    assert resp.status == HTTPStatus.METHOD_NOT_ALLOWED
+
+    # Check we did not forward command
+    assert len(aioclient_mock.mock_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("bad_path", "expected_status"),
+    [
+        # Caught by bullshit filter
+        ("app/%252E./entrypoint.js", HTTPStatus.BAD_REQUEST),
+        # The .. is processed, making it an unauthenticated path
+        ("app/../entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        ("app/%2E%2E/entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        # Unauthenticated path
+        ("supervisor/info", HTTPStatus.UNAUTHORIZED),
+        ("supervisor/logs", HTTPStatus.UNAUTHORIZED),
+        ("addons/bl_b392/logs", HTTPStatus.UNAUTHORIZED),
+    ],
+)
+async def test_forward_request_onboarded_noauth_unallowed_paths(
+    hassio_noauth_client,
+    aioclient_mock: AiohttpClientMocker,
+    bad_path: str,
+    expected_status: int,
+) -> None:
+    """Test fetching normal path."""
+    resp = await hassio_noauth_client.get(f"/api/hassio/{bad_path}")
+
+    # Check we got right response
+    assert resp.status == expected_status
+    # Check we didn't forward command
+    assert len(aioclient_mock.mock_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("path", "authenticated"),
+    [
+        ("app/entrypoint.js", False),
+        ("addons/bl_b392/logo", False),
+        ("addons/bl_b392/icon", False),
+        ("backups/1234abcd/info", True),
+    ],
+)
+async def test_forward_request_not_onboarded_get(
+    hassio_noauth_client,
+    aioclient_mock: AiohttpClientMocker,
+    path: str,
+    authenticated: bool,
+    mock_not_onboarded,
+) -> None:
+    """Test fetching normal path."""
+    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")
+
+    resp = await hassio_noauth_client.get(f"/api/hassio/{path}")
 
     # Check we got right response
     assert resp.status == HTTPStatus.OK
@@ -100,61 +193,224 @@ async def test_forward_request_no_auth_for_icon(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
+    expected_headers = {
+        "X-Hass-Source": "core.http",
+    }
+    if authenticated:
+        expected_headers["Authorization"] = "Bearer 123456"
+
+    assert aioclient_mock.mock_calls[0][3] == expected_headers
 
 
-async def test_forward_log_request(
-    hassio_client, aioclient_mock: AiohttpClientMocker
+@pytest.mark.parametrize(
+    "path",
+    [
+        "backups/new/upload",
+        "backups/1234abcd/restore/full",
+        "backups/1234abcd/restore/partial",
+    ],
+)
+async def test_forward_request_not_onboarded_post(
+    hassio_noauth_client,
+    aioclient_mock: AiohttpClientMocker,
+    path: str,
+    mock_not_onboarded,
 ) -> None:
-    """Test fetching normal log path doesn't remove ANSI color escape codes."""
-    aioclient_mock.get("http://127.0.0.1/beer/logs", text="\033[32mresponse\033[0m")
+    """Test fetching normal path."""
+    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")
 
-    resp = await hassio_client.get("/api/hassio/beer/logs")
+    resp = await hassio_noauth_client.get(f"/api/hassio/{path}")
 
     # Check we got right response
     assert resp.status == HTTPStatus.OK
     body = await resp.text()
-    assert body == "\033[32mresponse\033[0m"
+    assert body == "response"
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
+    # We only expect a single header.
+    assert aioclient_mock.mock_calls[0][3] == {
+        "X-Hass-Source": "core.http",
+        "Authorization": "Bearer 123456",
+    }
+
+
+@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE", "RANDOM"])
+async def test_forward_request_not_onboarded_unallowed_methods(
+    hassio_noauth_client, aioclient_mock: AiohttpClientMocker, method: str
+) -> None:
+    """Test fetching normal path."""
+    resp = await hassio_noauth_client.post("/api/hassio/app/entrypoint.js")
+
+    # Check we got right response
+    assert resp.status == HTTPStatus.METHOD_NOT_ALLOWED
+
+    # Check we did not forward command
+    assert len(aioclient_mock.mock_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("bad_path", "expected_status"),
+    [
+        # Caught by bullshit filter
+        ("app/%252E./entrypoint.js", HTTPStatus.BAD_REQUEST),
+        # The .. is processed, making it an unauthenticated path
+        ("app/../entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        ("app/%2E%2E/entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        # Unauthenticated path
+        ("supervisor/info", HTTPStatus.UNAUTHORIZED),
+        ("supervisor/logs", HTTPStatus.UNAUTHORIZED),
+        ("addons/bl_b392/logs", HTTPStatus.UNAUTHORIZED),
+    ],
+)
+async def test_forward_request_not_onboarded_unallowed_paths(
+    hassio_noauth_client,
+    aioclient_mock: AiohttpClientMocker,
+    bad_path: str,
+    expected_status: int,
+    mock_not_onboarded,
+) -> None:
+    """Test fetching normal path."""
+    resp = await hassio_noauth_client.get(f"/api/hassio/{bad_path}")
+
+    # Check we got right response
+    assert resp.status == expected_status
+    # Check we didn't forward command
+    assert len(aioclient_mock.mock_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("path", "authenticated"),
+    [
+        ("app/entrypoint.js", False),
+        ("addons/bl_b392/logo", False),
+        ("addons/bl_b392/icon", False),
+        ("backups/1234abcd/info", True),
+        ("supervisor/logs", True),
+        ("addons/bl_b392/logs", True),
+    ],
+)
+async def test_forward_request_admin_get(
+    hassio_client,
+    aioclient_mock: AiohttpClientMocker,
+    path: str,
+    authenticated: bool,
+) -> None:
+    """Test fetching normal path."""
+    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")
+
+    resp = await hassio_client.get(f"/api/hassio/{path}")
+
+    # Check we got right response
+    assert resp.status == HTTPStatus.OK
+    body = await resp.text()
+    assert body == "response"
+
+    # Check we forwarded command
+    assert len(aioclient_mock.mock_calls) == 1
+    expected_headers = {
+        "X-Hass-Source": "core.http",
+    }
+    if authenticated:
+        expected_headers["Authorization"] = "Bearer 123456"
+
+    assert aioclient_mock.mock_calls[0][3] == expected_headers
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "backups/new/upload",
+        "backups/1234abcd/restore/full",
+        "backups/1234abcd/restore/partial",
+    ],
+)
+async def test_forward_request_admin_post(
+    hassio_client,
+    aioclient_mock: AiohttpClientMocker,
+    path: str,
+) -> None:
+    """Test fetching normal path."""
+    aioclient_mock.get(f"http://127.0.0.1/{path}", text="response")
+
+    resp = await hassio_client.get(f"/api/hassio/{path}")
+
+    # Check we got right response
+    assert resp.status == HTTPStatus.OK
+    body = await resp.text()
+    assert body == "response"
+
+    # Check we forwarded command
+    assert len(aioclient_mock.mock_calls) == 1
+    # We only expect a single header.
+    assert aioclient_mock.mock_calls[0][3] == {
+        "X-Hass-Source": "core.http",
+        "Authorization": "Bearer 123456",
+    }
+
+
+@pytest.mark.parametrize("method", ["POST", "PUT", "DELETE", "RANDOM"])
+async def test_forward_request_admin_unallowed_methods(
+    hassio_client, aioclient_mock: AiohttpClientMocker, method: str
+) -> None:
+    """Test fetching normal path."""
+    resp = await hassio_client.post("/api/hassio/app/entrypoint.js")
+
+    # Check we got right response
+    assert resp.status == HTTPStatus.METHOD_NOT_ALLOWED
+
+    # Check we did not forward command
+    assert len(aioclient_mock.mock_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("bad_path", "expected_status"),
+    [
+        # Caught by bullshit filter
+        ("app/%252E./entrypoint.js", HTTPStatus.BAD_REQUEST),
+        # The .. is processed, making it an unauthenticated path
+        ("app/../entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        ("app/%2E%2E/entrypoint.js", HTTPStatus.UNAUTHORIZED),
+        # Unauthenticated path
+        ("supervisor/info", HTTPStatus.UNAUTHORIZED),
+    ],
+)
+async def test_forward_request_admin_unallowed_paths(
+    hassio_client,
+    aioclient_mock: AiohttpClientMocker,
+    bad_path: str,
+    expected_status: int,
+) -> None:
+    """Test fetching normal path."""
+    resp = await hassio_client.get(f"/api/hassio/{bad_path}")
+
+    # Check we got right response
+    assert resp.status == expected_status
+    # Check we didn't forward command
+    assert len(aioclient_mock.mock_calls) == 0
 
 
 async def test_bad_gateway_when_cannot_find_supervisor(
     hassio_client, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test we get a bad gateway error if we can't find supervisor."""
-    aioclient_mock.get("http://127.0.0.1/addons/test/info", exc=asyncio.TimeoutError)
+    aioclient_mock.get("http://127.0.0.1/app/entrypoint.js", exc=asyncio.TimeoutError)
 
-    resp = await hassio_client.get("/api/hassio/addons/test/info")
+    resp = await hassio_client.get("/api/hassio/app/entrypoint.js")
     assert resp.status == HTTPStatus.BAD_GATEWAY
 
 
-async def test_forwarding_user_info(
-    hassio_client, hass_admin_user: MockUser, aioclient_mock: AiohttpClientMocker
-) -> None:
-    """Test that we forward user info correctly."""
-    aioclient_mock.get("http://127.0.0.1/hello")
-
-    resp = await hassio_client.get("/api/hassio/hello")
-
-    # Check we got right response
-    assert resp.status == HTTPStatus.OK
-
-    assert len(aioclient_mock.mock_calls) == 1
-
-    req_headers = aioclient_mock.mock_calls[0][-1]
-    assert req_headers["X-Hass-User-ID"] == hass_admin_user.id
-    assert req_headers["X-Hass-Is-Admin"] == "1"
-
-
 async def test_backup_upload_headers(
-    hassio_client, aioclient_mock: AiohttpClientMocker, caplog: pytest.LogCaptureFixture
+    hassio_client,
+    aioclient_mock: AiohttpClientMocker,
+    caplog: pytest.LogCaptureFixture,
+    mock_not_onboarded,
 ) -> None:
     """Test that we forward the full header for backup upload."""
     content_type = "multipart/form-data; boundary='--webkit'"
-    aioclient_mock.get("http://127.0.0.1/backups/new/upload")
+    aioclient_mock.post("http://127.0.0.1/backups/new/upload")
 
-    resp = await hassio_client.get(
+    resp = await hassio_client.post(
         "/api/hassio/backups/new/upload", headers={"Content-Type": content_type}
     )
 
@@ -168,19 +424,19 @@ async def test_backup_upload_headers(
 
 
 async def test_backup_download_headers(
-    hassio_client, aioclient_mock: AiohttpClientMocker
+    hassio_client, aioclient_mock: AiohttpClientMocker, mock_not_onboarded
 ) -> None:
     """Test that we forward the full header for backup download."""
     content_disposition = "attachment; filename=test.tar"
     aioclient_mock.get(
-        "http://127.0.0.1/backups/slug/download",
+        "http://127.0.0.1/backups/1234abcd/download",
         headers={
             "Content-Length": "50000000",
             "Content-Disposition": content_disposition,
         },
     )
 
-    resp = await hassio_client.get("/api/hassio/backups/slug/download")
+    resp = await hassio_client.get("/api/hassio/backups/1234abcd/download")
 
     # Check we got right response
     assert resp.status == HTTPStatus.OK
@@ -190,21 +446,10 @@ async def test_backup_download_headers(
     assert resp.headers["Content-Disposition"] == content_disposition
 
 
-def test_need_auth(hass: HomeAssistant) -> None:
-    """Test if the requested path needs authentication."""
-    assert not _need_auth(hass, "addons/test/logo")
-    assert _need_auth(hass, "backups/new/upload")
-    assert _need_auth(hass, "supervisor/logs")
-
-    hass.data["onboarding"] = False
-    assert not _need_auth(hass, "backups/new/upload")
-    assert not _need_auth(hass, "supervisor/logs")
-
-
 async def test_stream(hassio_client, aioclient_mock: AiohttpClientMocker) -> None:
     """Verify that the request is a stream."""
-    aioclient_mock.get("http://127.0.0.1/test")
-    await hassio_client.get("/api/hassio/test", data="test")
+    aioclient_mock.get("http://127.0.0.1/app/entrypoint.js")
+    await hassio_client.get("/api/hassio/app/entrypoint.js", data="test")
     assert isinstance(aioclient_mock.mock_calls[-1][2], StreamReader)
 
 
diff --git a/tests/components/hassio/test_ingress.py b/tests/components/hassio/test_ingress.py
index 52ca535516a..67548a19c2c 100644
--- a/tests/components/hassio/test_ingress.py
+++ b/tests/components/hassio/test_ingress.py
@@ -21,7 +21,7 @@ from tests.test_util.aiohttp import AiohttpClientMocker
     ],
 )
 async def test_ingress_request_get(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+    hassio_noauth_client, build_type, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test no auth needed for ."""
     aioclient_mock.get(
@@ -29,7 +29,7 @@ async def test_ingress_request_get(
         text="test",
     )
 
-    resp = await hassio_client.get(
+    resp = await hassio_noauth_client.get(
         f"/api/hassio_ingress/{build_type[0]}/{build_type[1]}",
         headers={"X-Test-Header": "beer"},
     )
@@ -41,7 +41,8 @@ async def test_ingress_request_get(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
-    assert aioclient_mock.mock_calls[-1][3][X_AUTH_TOKEN] == "123456"
+    assert X_AUTH_TOKEN not in aioclient_mock.mock_calls[-1][3]
+    assert aioclient_mock.mock_calls[-1][3]["X-Hass-Source"] == "core.ingress"
     assert (
         aioclient_mock.mock_calls[-1][3]["X-Ingress-Path"]
         == f"/api/hassio_ingress/{build_type[0]}"
@@ -63,7 +64,7 @@ async def test_ingress_request_get(
     ],
 )
 async def test_ingress_request_post(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+    hassio_noauth_client, build_type, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test no auth needed for ."""
     aioclient_mock.post(
@@ -71,7 +72,7 @@ async def test_ingress_request_post(
         text="test",
     )
 
-    resp = await hassio_client.post(
+    resp = await hassio_noauth_client.post(
         f"/api/hassio_ingress/{build_type[0]}/{build_type[1]}",
         headers={"X-Test-Header": "beer"},
     )
@@ -83,7 +84,8 @@ async def test_ingress_request_post(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
-    assert aioclient_mock.mock_calls[-1][3][X_AUTH_TOKEN] == "123456"
+    assert X_AUTH_TOKEN not in aioclient_mock.mock_calls[-1][3]
+    assert aioclient_mock.mock_calls[-1][3]["X-Hass-Source"] == "core.ingress"
     assert (
         aioclient_mock.mock_calls[-1][3]["X-Ingress-Path"]
         == f"/api/hassio_ingress/{build_type[0]}"
@@ -105,7 +107,7 @@ async def test_ingress_request_post(
     ],
 )
 async def test_ingress_request_put(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+    hassio_noauth_client, build_type, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test no auth needed for ."""
     aioclient_mock.put(
@@ -113,7 +115,7 @@ async def test_ingress_request_put(
         text="test",
     )
 
-    resp = await hassio_client.put(
+    resp = await hassio_noauth_client.put(
         f"/api/hassio_ingress/{build_type[0]}/{build_type[1]}",
         headers={"X-Test-Header": "beer"},
     )
@@ -125,7 +127,8 @@ async def test_ingress_request_put(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
-    assert aioclient_mock.mock_calls[-1][3][X_AUTH_TOKEN] == "123456"
+    assert X_AUTH_TOKEN not in aioclient_mock.mock_calls[-1][3]
+    assert aioclient_mock.mock_calls[-1][3]["X-Hass-Source"] == "core.ingress"
     assert (
         aioclient_mock.mock_calls[-1][3]["X-Ingress-Path"]
         == f"/api/hassio_ingress/{build_type[0]}"
@@ -147,7 +150,7 @@ async def test_ingress_request_put(
     ],
 )
 async def test_ingress_request_delete(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+    hassio_noauth_client, build_type, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test no auth needed for ."""
     aioclient_mock.delete(
@@ -155,7 +158,7 @@ async def test_ingress_request_delete(
         text="test",
     )
 
-    resp = await hassio_client.delete(
+    resp = await hassio_noauth_client.delete(
         f"/api/hassio_ingress/{build_type[0]}/{build_type[1]}",
         headers={"X-Test-Header": "beer"},
     )
@@ -167,7 +170,8 @@ async def test_ingress_request_delete(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
-    assert aioclient_mock.mock_calls[-1][3][X_AUTH_TOKEN] == "123456"
+    assert X_AUTH_TOKEN not in aioclient_mock.mock_calls[-1][3]
+    assert aioclient_mock.mock_calls[-1][3]["X-Hass-Source"] == "core.ingress"
     assert (
         aioclient_mock.mock_calls[-1][3]["X-Ingress-Path"]
         == f"/api/hassio_ingress/{build_type[0]}"
@@ -189,7 +193,7 @@ async def test_ingress_request_delete(
     ],
 )
 async def test_ingress_request_patch(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+    hassio_noauth_client, build_type, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test no auth needed for ."""
     aioclient_mock.patch(
@@ -197,7 +201,7 @@ async def test_ingress_request_patch(
         text="test",
     )
 
-    resp = await hassio_client.patch(
+    resp = await hassio_noauth_client.patch(
         f"/api/hassio_ingress/{build_type[0]}/{build_type[1]}",
         headers={"X-Test-Header": "beer"},
     )
@@ -209,7 +213,8 @@ async def test_ingress_request_patch(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
-    assert aioclient_mock.mock_calls[-1][3][X_AUTH_TOKEN] == "123456"
+    assert X_AUTH_TOKEN not in aioclient_mock.mock_calls[-1][3]
+    assert aioclient_mock.mock_calls[-1][3]["X-Hass-Source"] == "core.ingress"
     assert (
         aioclient_mock.mock_calls[-1][3]["X-Ingress-Path"]
         == f"/api/hassio_ingress/{build_type[0]}"
@@ -231,7 +236,7 @@ async def test_ingress_request_patch(
     ],
 )
 async def test_ingress_request_options(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+    hassio_noauth_client, build_type, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test no auth needed for ."""
     aioclient_mock.options(
@@ -239,7 +244,7 @@ async def test_ingress_request_options(
         text="test",
     )
 
-    resp = await hassio_client.options(
+    resp = await hassio_noauth_client.options(
         f"/api/hassio_ingress/{build_type[0]}/{build_type[1]}",
         headers={"X-Test-Header": "beer"},
     )
@@ -251,7 +256,8 @@ async def test_ingress_request_options(
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
-    assert aioclient_mock.mock_calls[-1][3][X_AUTH_TOKEN] == "123456"
+    assert X_AUTH_TOKEN not in aioclient_mock.mock_calls[-1][3]
+    assert aioclient_mock.mock_calls[-1][3]["X-Hass-Source"] == "core.ingress"
     assert (
         aioclient_mock.mock_calls[-1][3]["X-Ingress-Path"]
         == f"/api/hassio_ingress/{build_type[0]}"
@@ -273,20 +279,21 @@ async def test_ingress_request_options(
     ],
 )
 async def test_ingress_websocket(
-    hassio_client, build_type, aioclient_mock: AiohttpClientMocker
+    hassio_noauth_client, build_type, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test no auth needed for ."""
     aioclient_mock.get(f"http://127.0.0.1/ingress/{build_type[0]}/{build_type[1]}")
 
     # Ignore error because we can setup a full IO infrastructure
-    await hassio_client.ws_connect(
+    await hassio_noauth_client.ws_connect(
         f"/api/hassio_ingress/{build_type[0]}/{build_type[1]}",
         headers={"X-Test-Header": "beer"},
     )
 
     # Check we forwarded command
     assert len(aioclient_mock.mock_calls) == 1
-    assert aioclient_mock.mock_calls[-1][3][X_AUTH_TOKEN] == "123456"
+    assert X_AUTH_TOKEN not in aioclient_mock.mock_calls[-1][3]
+    assert aioclient_mock.mock_calls[-1][3]["X-Hass-Source"] == "core.ingress"
     assert (
         aioclient_mock.mock_calls[-1][3]["X-Ingress-Path"]
         == f"/api/hassio_ingress/{build_type[0]}"
@@ -298,7 +305,9 @@ async def test_ingress_websocket(
 
 
 async def test_ingress_missing_peername(
-    hassio_client, aioclient_mock: AiohttpClientMocker, caplog: pytest.LogCaptureFixture
+    hassio_noauth_client,
+    aioclient_mock: AiohttpClientMocker,
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test hadnling of missing peername."""
     aioclient_mock.get(
@@ -314,7 +323,7 @@ async def test_ingress_missing_peername(
         return_value=MagicMock(),
     ) as transport_mock:
         transport_mock.get_extra_info = get_extra_info
-        resp = await hassio_client.get(
+        resp = await hassio_noauth_client.get(
             "/api/hassio_ingress/lorem/ipsum",
             headers={"X-Test-Header": "beer"},
         )
@@ -323,3 +332,19 @@ async def test_ingress_missing_peername(
 
     # Check we got right response
     assert resp.status == HTTPStatus.BAD_REQUEST
+
+
+async def test_forwarding_paths_as_requested(
+    hassio_noauth_client, aioclient_mock
+) -> None:
+    """Test incomnig URLs with double encoding go out as dobule encoded."""
+    # This double encoded string should be forwarded double-encoded too.
+    aioclient_mock.get(
+        "http://127.0.0.1/ingress/mock-token/hello/%252e./world",
+        text="test",
+    )
+
+    resp = await hassio_noauth_client.get(
+        "/api/hassio_ingress/mock-token/hello/%252e./world",
+    )
+    assert await resp.text() == "test"
diff --git a/tests/components/hassio/test_websocket_api.py b/tests/components/hassio/test_websocket_api.py
index 611ada61814..b2f9e06cb43 100644
--- a/tests/components/hassio/test_websocket_api.py
+++ b/tests/components/hassio/test_websocket_api.py
@@ -153,6 +153,11 @@ async def test_websocket_supervisor_api(
     msg = await websocket_client.receive_json()
     assert msg["result"]["version_latest"] == "1.0.0"
 
+    assert aioclient_mock.mock_calls[-1][3] == {
+        "X-Hass-Source": "core.websocket_api",
+        "Authorization": "Bearer 123456",
+    }
+
 
 async def test_websocket_supervisor_api_error(
     hassio_env,