Remove check for whether localhost connection is root (#14068)

We don't use this functionality internally and it's a potential security liability if someone decides
to set up their own internal proxy to middlewared socket that's running as root, or if an application
allows root account and has access to host networking.

src/middlewared/middlewared/auth.py

@@ -59,10 +59,6 @@ class UnixSocketSessionManagerCredentials(UserSessionManagerCredentials):
     pass
 
 
-class RootTcpSocketSessionManagerCredentials(UserSessionManagerCredentials):
-    pass
-
-
 class LoginPasswordSessionManagerCredentials(UserSessionManagerCredentials):
     pass
 

src/middlewared/middlewared/plugins/auth.py

@@ -8,17 +8,15 @@
 import psutil
 
 from middlewared.auth import (SessionManagerCredentials, UserSessionManagerCredentials,
-                              UnixSocketSessionManagerCredentials, RootTcpSocketSessionManagerCredentials,
-                              LoginPasswordSessionManagerCredentials, ApiKeySessionManagerCredentials,
-                              TrueNasNodeSessionManagerCredentials)
+                              UnixSocketSessionManagerCredentials, LoginPasswordSessionManagerCredentials,
+                              ApiKeySessionManagerCredentials, TrueNasNodeSessionManagerCredentials)
 from middlewared.schema import accepts, Any, Bool, Datetime, Dict, Int, List, Password, Patch, Ref, returns, Str
 from middlewared.service import (
     Service, filterable, filterable_returns, filter_list, no_auth_required, no_authz_required,
     pass_app, private, cli_private, CallError,
 )
 from middlewared.service_exception import MatchNotFound
 import middlewared.sqlalchemy as sa
-from middlewared.utils.nginx import get_peer_process
 from middlewared.utils.origin import UnixSocketOrigin, TCPIPOrigin
 from middlewared.utils.crypto import generate_token
 
@@ -210,10 +208,7 @@ def is_internal_session(session):
     if isinstance(session.app.origin, UnixSocketOrigin) and session.app.origin.uid == 0:
         return True
 
-    if isinstance(session.app.authenticated_credentials, (
-        RootTcpSocketSessionManagerCredentials,
-        TrueNasNodeSessionManagerCredentials,
-    )):
+    if isinstance(session.app.authenticated_credentials, TrueNasNodeSessionManagerCredentials):
         return True
 
     return False
@@ -673,22 +668,6 @@ async def check_permission(middleware, app):
         await AuthService.session_manager.login(app, UnixSocketSessionManagerCredentials(user))
         return
 
-    if isinstance(origin, TCPIPOrigin):
-        if not (origin.addr.startswith('127.') or origin.addr == '::1'):
-            return
-
-        # This is an expensive operation, but it is only performed for localhost TCP connections which are rare
-        if process := await middleware.run_in_thread(get_peer_process, origin.addr, origin.port):
-            try:
-                euid = (await middleware.run_in_thread(process.uids)).effective
-            except psutil.NoSuchProcess:
-                pass
-            else:
-                if euid == 0:
-                    user = await middleware.call('auth.authenticate_root')
-                    await AuthService.session_manager.login(app, RootTcpSocketSessionManagerCredentials(user))
-                    return
-
 
 def setup(middleware):
     middleware.event_register('auth.sessions', 'Notification of new and removed sessions.')

src/middlewared/middlewared/utils/nginx.py

@@ -3,15 +3,6 @@
 from psutil._common import addr
 
 
-def get_peer_process(remote_addr, remote_port):
-    for connection in psutil.net_connections(kind='tcp'):
-        if connection.laddr == addr(remote_addr, remote_port):
-            try:
-                return psutil.Process(connection.pid)
-            except psutil.ProcessNotFound:
-                return None
-
-
 def get_remote_addr_port(request):
     try:
         remote_addr, remote_port = request.transport.get_extra_info("peername")

src/middlewared/middlewared/utils/privilege.py

@@ -1,6 +1,5 @@
 import enum
-from middlewared.auth import (RootTcpSocketSessionManagerCredentials,
-                              TrueNasNodeSessionManagerCredentials)
+from middlewared.auth import TrueNasNodeSessionManagerCredentials
 from middlewared.role import ROLES
 
 
@@ -29,10 +28,7 @@ def credential_has_full_admin(credential: object) -> bool:
     if credential.is_user_session and 'FULL_ADMIN' in credential.user['privilege']['roles']:
         return True
 
-    if isinstance(credential, (
-        RootTcpSocketSessionManagerCredentials,
-        TrueNasNodeSessionManagerCredentials
-    )):
+    if isinstance(credential, TrueNasNodeSessionManagerCredentials):
         return True
 
     if credential.allowlist is None:

tests/api2/test_localhost_ws_auth.py

@@ -0,0 +1,11 @@
+from middlewared.test.integration.utils import ssh
+
+
+def test__authentication_required_localhost():
+    cmd = 'midclt -u ws://localhost/websocket call user.query'
+    resp = ssh(cmd, check=False, complete_response=True)
+
+    assert not resp['result']
+
+    assert 'Not authenticated' in resp['stderr']
+