@@ -9,6 +9,7 @@
 from aiohttp import ClientTimeout, ClientWSTimeout, WSCloseCode, hdrs, web
 from aiohttp.web_exceptions import (
     HTTPBadGateway,
+    HTTPForbidden,
     HTTPServiceUnavailable,
     HTTPUnauthorized,
 )
@@ -35,7 +36,7 @@
     IngressSessionData,
 )
 from ..coresys import CoreSysAttributes
-from ..exceptions import HomeAssistantAPIError
+from ..exceptions import APIError, HomeAssistantAPIError
 from .const import COOKIE_INGRESS, WEBSOCKETS
 from .utils import api_process, api_validate, require_home_assistant, stop_on_disconnect
 
@@ -48,7 +49,7 @@
 """Expected optional payload of create session request"""
 SCHEMA_INGRESS_CREATE_SESSION_DATA = vol.Schema(
     {
-        vol.Optional(ATTR_SESSION_DATA_USER_ID): str,
+        vol.Required(ATTR_SESSION_DATA_USER_ID): str,
     }
 )
 
@@ -113,19 +114,21 @@ async def panels(self, request: web.Request) -> dict[str, Any]:
     @require_home_assistant
     async def create_session(self, request: web.Request) -> dict[str, Any]:
         """Create a new session."""
-        schema_ingress_config_session_data = await api_validate(
-            SCHEMA_INGRESS_CREATE_SESSION_DATA, request
-        )
-        data: IngressSessionData | None = None
+        body = await api_validate(SCHEMA_INGRESS_CREATE_SESSION_DATA, request)
+        user_id = body[ATTR_SESSION_DATA_USER_ID]
 
-        if ATTR_SESSION_DATA_USER_ID in schema_ingress_config_session_data:
-            user = await self._find_user_by_id(
-                schema_ingress_config_session_data[ATTR_SESSION_DATA_USER_ID]
-            )
-            if user:
-                data = IngressSessionData(user)
+        try:
+            user = await self._find_user_by_id(user_id)
+        except HomeAssistantAPIError as err:
+            _LOGGER.warning("Could not fetch list of users: %s", err)
+            # Keep the ID so the user can be resolved on first use
+            user = HomeAssistantUser(user_id)
+        else:
+            if user is None or not user.is_active:
+                raise APIError(f"User {user_id} does not exist or is inactive")
+            self.sys_ingress.users[user.id] = user
 
-        session = self.sys_ingress.create_session(data)
+        session = self.sys_ingress.create_session(IngressSessionData(user))
         return {ATTR_SESSION: session}
 
     @api_process
@@ -154,13 +157,21 @@ async def handler(
         app = self._extract_app(request)
         path = request.match_info.get("path", "")
         session_data = self.sys_ingress.get_session_data(session)
+        user = await self._get_user(session, session_data) if session_data else None
+        if app.panel_admin and (user is None or not user.is_admin):
+            _LOGGER.warning(
+                "Ingress access to %s denied for non-admin user %s",
+                app.slug,
+                user.id if user else None,
+            )
+            raise HTTPForbidden
         try:
             # Websocket
             if _is_websocket(request):
-                return await self._handle_websocket(request, app, path, session_data)
+                return await self._handle_websocket(request, app, path, user)
 
             # Request
-            return await self._handle_request(request, app, path, session_data)
+            return await self._handle_request(request, app, path, user)
 
         except aiohttp.ClientError as err:
             _LOGGER.error("Ingress error: %s", err)
@@ -172,7 +183,7 @@ async def _handle_websocket(
         request: web.Request,
         app: App,
         path: str,
-        session_data: IngressSessionData | None,
+        user: HomeAssistantUser | None,
     ) -> web.WebSocketResponse:
         """Ingress route for websocket."""
         if hdrs.SEC_WEBSOCKET_PROTOCOL in request.headers:
@@ -198,7 +209,7 @@ async def _handle_websocket(
 
         # Preparing
         url = self._create_url(app, path)
-        source_header = _init_header(request, app, session_data)
+        source_header = _init_header(request, app, user)
 
         # Support GET query
         if request.query_string:
@@ -238,11 +249,11 @@ async def _handle_request(
         request: web.Request,
         app: App,
         path: str,
-        session_data: IngressSessionData | None,
+        user: HomeAssistantUser | None,
     ) -> web.Response | web.StreamResponse:
         """Ingress route for request."""
         url = self._create_url(app, path)
-        source_header = _init_header(request, app, session_data)
+        source_header = _init_header(request, app, user)
 
         # Passing the raw stream breaks requests for some webservers
         # since we just need it for POST requests really, for all other methods
@@ -320,28 +331,54 @@ async def _handle_request(
             return response
 
     async def _find_user_by_id(self, user_id: str) -> HomeAssistantUser | None:
-        """Find user object by the user's ID."""
+        """Find user object by the user's ID.
+
+        Raises HomeAssistantAPIError if the user list cannot be fetched.
+        """
+        users = await self.sys_homeassistant.list_users()
+        return next((user for user in users if user.id == user_id), None)
+
+    async def _get_user(
+        self, session: str, session_data: IngressSessionData
+    ) -> HomeAssistantUser:
+        """Return the user record of a session, resolved from Core once.
+
+        While Core is unreachable the record stored with the session is
+        returned, which denies admin access without caching. Sessions of
+        removed or deactivated users are invalidated.
+        """
+        if user := self.sys_ingress.users.get(session_data.user.id):
+            return user
+
         try:
-            users = await self.sys_homeassistant.list_users()
+            user = await self._find_user_by_id(session_data.user.id)
         except HomeAssistantAPIError as err:
             _LOGGER.warning("Could not fetch list of users: %s", err)
-            return None
+            return session_data.user
 
-        return next((user for user in users if user.id == user_id), None)
+        if user is None or not user.is_active:
+            _LOGGER.warning(
+                "Ingress session of removed or inactive user %s", session_data.user.id
+            )
+            self.sys_ingress.remove_session(session)
+            raise HTTPUnauthorized
+
+        self.sys_ingress.users[user.id] = user
+        return user
 
 
 def _init_header(
-    request: web.Request, app: App, session_data: IngressSessionData | None
+    request: web.Request, app: App, user: HomeAssistantUser | None
 ) -> CIMultiDict[str]:
     """Create initial header."""
     headers = CIMultiDict[str]()
 
-    if session_data is not None:
-        headers[HEADER_REMOTE_USER_ID] = session_data.user.id
-        if session_data.user.username is not None:
-            headers[HEADER_REMOTE_USER_NAME] = session_data.user.username
-        if session_data.user.name is not None:
-            headers[HEADER_REMOTE_USER_DISPLAY_NAME] = session_data.user.name
+    if user is not None:
+        headers[HEADER_REMOTE_USER_ID] = user.id
+        if user.username is not None:
+            headers[HEADER_REMOTE_USER_NAME] = user.username
+        if user.name is not None:
+            headers[HEADER_REMOTE_USER_DISPLAY_NAME] = user.name
 
     # filter flags
     for name, value in request.headers.items():
 
@@ -609,6 +609,9 @@ class FeatureFlag(StrEnum):
     SUPERVISOR_WEBSOCKET_V2_API = "supervisor_websocket_v2_api"
 
 
+HOMEASSISTANT_GROUP_ID_ADMIN = "system-admin"
+
+
 @dataclass
 class HomeAssistantUser:
     """A Home Assistant Core user.
@@ -627,6 +630,13 @@ class HomeAssistantUser:
     system_generated: bool = False
     group_ids: list[str] | None = None
 
+    @property
+    def is_admin(self) -> bool:
+        """Return if the user is an admin, mirroring Core's User.is_admin."""
+        return self.is_owner or (
+            self.is_active and HOMEASSISTANT_GROUP_ID_ADMIN in (self.group_ids or [])
+        )
+
     @classmethod
     def from_dict(cls, data: Mapping[str, Any]) -> Self:
         """Return object from dictionary representation."""
 
@@ -13,6 +13,7 @@
     FILE_HASSIO_INGRESS,
     INGRESS_DYNAMIC_PORT_MAX,
     INGRESS_DYNAMIC_PORT_MIN,
+    HomeAssistantUser,
     IngressSessionData,
     IngressSessionDataDict,
 )
@@ -34,6 +35,9 @@ def __init__(self, coresys: CoreSys):
         super().__init__(FILE_HASSIO_INGRESS, SCHEMA_INGRESS_CONFIG)
         self.coresys: CoreSys = coresys
         self.tokens: dict[str, str] = {}
+        # User records by ID, resolved from Core and not persisted so a
+        # restart picks up user changes. Only known users are ever added.
+        self.users: dict[str, HomeAssistantUser] = {}
 
     def get(self, token: str) -> App | None:
         """Return app they have this ingress token."""
@@ -105,10 +109,12 @@ def _cleanup_sessions(self) -> None:
             if valid_dt < now:
                 continue
 
-            # Is valid
+            # Sessions without user data predate Core 2023.9 and cannot be authorized
+            if not (session_data := self.sessions_data.get(session)):
+                continue
+
             sessions[session] = valid
-            if session_data := self.sessions_data.get(session):
-                sessions_data[session] = session_data
+            sessions_data[session] = session_data
 
         # Write back
         self.sessions.clear()
@@ -136,6 +142,11 @@ def create_session(self, data: IngressSessionData | None = None) -> str:
 
         return session
 
+    def remove_session(self, session: str) -> None:
+        """Invalidate a session."""
+        self.sessions.pop(session, None)
+        self.sessions_data.pop(session, None)
+
     def validate_session(self, session: str) -> bool:
         """Return True if session valid and make it longer valid."""
         if session not in self.sessions: