NAS-130198 / 24.04.3 / Remove check for whether localhost connection is root (by anodos325)

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']
+