feat(acl): enhance ACL support with GUI synchronization and credentia…

.github/actions/bec_e2e_install/action.yml

@@ -77,5 +77,6 @@ runs:
         pip install -e ../bec_testing_plugin
         podman pod create --network=host local_bec
         python ./bec_ipython_client/tests/end-2-end/_ensure_requirements_container.py
-        coverage run --branch --source=./bec_server/bec_server,./bec_lib/bec_lib,./bec_ipython_client/bec_ipython_client, --omit=*/bec_server/scan_server/scan_plugins/*,*/bec_ipython_client/bec_ipython_client/plugins/*,*/bec_ipython_client/scripts/*,*/bec_lib/bec_lib/tests/* -m pytest -v --files-path ./ --start-servers --random-order  ./bec_ipython_client/tests/end-2-end/
+        coverage run --branch --source=./bec_server/bec_server,./bec_lib/bec_lib,./bec_ipython_client/bec_ipython_client, --omit=*/bec_server/scan_server/scan_plugins/*,*/bec_ipython_client/bec_ipython_client/plugins/*,*/bec_lib/bec_lib/tests/* -m pytest -v --files-path ./ --start-servers --flush-redis --random-order ./bec_ipython_client/tests/end-2-end/test_acl_e2e.py
+        coverage run --append --branch --source=./bec_server/bec_server,./bec_lib/bec_lib,./bec_ipython_client/bec_ipython_client, --omit=*/bec_server/scan_server/scan_plugins/*,*/bec_ipython_client/bec_ipython_client/plugins/*,*/bec_lib/bec_lib/tests/* -m pytest -v --files-path ./ --start-servers --random-order --ignore=./bec_ipython_client/tests/end-2-end/test_acl_e2e.py ./bec_ipython_client/tests/end-2-end/
         coverage xml -o coverage-e2e.xml

bec_ipython_client/tests/end-2-end/test_acl_e2e.py

@@ -0,0 +1,140 @@
+import shutil
+
+import pytest
+import redis.exceptions
+
+from bec_lib import messages
+from bec_lib.client import BECClient
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.redis_connector import RedisConnector
+from bec_lib.service_config import ServiceConfig, ServiceConfigModel
+from bec_lib.tests.utils import wait_for_empty_queue
+from bec_lib.utils.user_acls_test import BECAccessDemo
+from bec_server.bec_server_utils.service_handler import ServiceHandler
+
+
+@pytest.fixture
+def acl_enabled_bec_services(
+    request,
+    bec_files_path,
+    bec_services_config_file_path,
+    bec_test_config_file_path,
+    bec_redis_host_port,
+    test_config_yaml_file_path,
+):
+    if not request.config.getoption("--start-servers") or not request.config.getoption(
+        "--flush-redis"
+    ):
+        pytest.skip("ACL e2e test mutates Redis ACLs and requires isolated pytest BEC services.")
+
+    redis_host, redis_port = bec_redis_host_port
+    redis_url = f"{redis_host}:{redis_port}"
+    acl_env_file = bec_services_config_file_path.parent / ".bec_acl.env"
+    acl_env_file.write_text("REDIS_USER=admin\nREDIS_PASSWORD=admin\n", encoding="utf-8")
+    shutil.copyfile(test_config_yaml_file_path, bec_test_config_file_path)
+
+    service_config = ServiceConfigModel(
+        redis={"host": redis_host, "port": redis_port},
+        file_writer={"base_path": str(bec_services_config_file_path.parent)},
+        acl={"env_file": str(acl_env_file)},
+    )
+    bec_services_config_file_path.write_text(
+        service_config.model_dump_json(indent=4), encoding="utf-8"
+    )
+
+    service_handler = ServiceHandler(
+        bec_path=bec_files_path, config_path=bec_services_config_file_path, interface="subprocess"
+    )
+    processes = None
+    try:
+        acl_connector = RedisConnector(redis_url)
+        try:
+            access_control = BECAccessDemo(acl_connector)
+            access_control.reset()
+            access_control.add_user()
+            access_control.add_admin()
+            access_control.set_default_limited(True)
+        finally:
+            acl_connector.shutdown()
+
+        processes = service_handler.start()
+        yield bec_services_config_file_path
+    finally:
+        if processes is not None:
+            service_handler.stop(processes)
+        restore_connector = RedisConnector(redis_url)
+        try:
+            BECAccessDemo(restore_connector).reset()
+        finally:
+            restore_connector.shutdown()
+
+
+@pytest.mark.timeout(120)
+def test_acl_admin_server_allows_default_user_scan(acl_enabled_bec_services):
+    admin_config = ServiceConfig(acl_enabled_bec_services)
+    admin_client = BECClient(admin_config, RedisConnector, forced=True, wait_for_server=True)
+    admin_client.start()
+    try:
+        admin_client.config.load_demo_config(force=True)
+    finally:
+        admin_client.shutdown()
+        admin_client._client._reset_singleton()
+
+    user_config = ServiceConfig(acl_enabled_bec_services, acl={"env_file": "", "user": "user"})
+    user_client = BECClient(user_config, RedisConnector, forced=True, wait_for_server=True)
+    user_client.start()
+    try:
+        user_client.queue.request_queue_reset()
+        user_client.queue.request_scan_continuation()
+        wait_for_empty_queue(user_client)
+        assert user_client.username == "user"
+
+        dev = user_client.device_manager.devices
+        status = user_client.scans.line_scan(
+            dev.samx, -0.1, 0.1, steps=3, exp_time=0.01, relative=False
+        )
+        status.wait(num_points=True, file_written=True)
+
+        assert status.scan.num_points == 3
+    finally:
+        user_client.shutdown()
+        user_client._client._reset_singleton()
+
+
+@pytest.mark.timeout(120)
+def test_acl_account_endpoint_allows_user_read_but_admin_only_write(acl_enabled_bec_services):
+    account_endpoint = MessageEndpoints.account()
+    initial_account_msg = messages.VariableMessage(value="initial-admin-account")
+    updated_account_msg = messages.VariableMessage(value="updated-admin-account")
+
+    user_config = ServiceConfig(acl_enabled_bec_services, acl={"env_file": "", "user": "user"})
+    user_client = BECClient(user_config, RedisConnector, forced=True, wait_for_server=True)
+    user_client.start()
+    try:
+        assert user_client.username == "user"
+        with user_client.acl.temporary_user(username="admin", token="admin"):
+            user_client._update_username()
+            assert user_client.username == "admin"
+            user_client.connector.xadd(account_endpoint, {"data": initial_account_msg})
+            assert user_client.connector.get_last(account_endpoint, "data") == initial_account_msg
+
+        user_client._update_username()
+        assert user_client.username == "user"
+        assert user_client.connector.get_last(account_endpoint, "data") == initial_account_msg
+        with pytest.raises(redis.exceptions.NoPermissionError):
+            user_client.connector.xadd(
+                account_endpoint, {"data": messages.VariableMessage(value="user-account")}
+            )
+
+        with user_client.acl.temporary_user(username="admin", token="admin"):
+            user_client._update_username()
+            assert user_client.username == "admin"
+            user_client.connector.xadd(account_endpoint, {"data": updated_account_msg})
+            assert user_client.connector.get_last(account_endpoint, "data") == updated_account_msg
+
+        user_client._update_username()
+        assert user_client.username == "user"
+        assert user_client.connector.get_last(account_endpoint, "data") == updated_account_msg
+    finally:
+        user_client.shutdown()
+        user_client._client._reset_singleton()

bec_lib/bec_lib/acl_login.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import os
+from collections.abc import Iterator
+from contextlib import contextmanager
 from functools import wraps
 from getpass import getpass
 from typing import TYPE_CHECKING, cast
@@ -13,6 +15,7 @@
 from bec_lib.endpoints import MessageEndpoints
 from bec_lib.logger import bec_logger
 from bec_lib.redis_connector import RedisConnector
+from bec_lib.service_config import ACLConfig
 from bec_lib.utils.import_utils import lazy_import_from
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -87,8 +90,35 @@ def login_with_token(self, *, username: str, token: str | None):
         """
         self.connector.authenticate(username=username, password=token)
 
+    @contextmanager
+    def temporary_user(self, *, username: str, token: str | None) -> Iterator[None]:
+        """
+        Temporarily authenticate as another ACL user and restore the current user on exit.
+        """
+        credentials = self.current_credentials()
+        self.login_with_token(username=username, token=token)
+        try:
+            yield
+        finally:
+            self.login_with_token(
+                username=credentials["username"] or "default", token=credentials.get("token")
+            )
+
+    def current_credentials(self) -> dict[str, str | None]:
+        """
+        Return the current Redis ACL credentials for forwarding to linked clients.
+        """
+        # pylint: disable=protected-access
+        conn_kwargs = (
+            self.connector._managed_connection._redis_conn.connection_pool.connection_kwargs
+        )
+        token = conn_kwargs.get("password")
+        if token == "null":
+            token = None
+        return {"username": self.connector.username, "token": token}
+
     def _bec_service_login(
-        self, prompt_for_acl: bool = False, acl_config: dict | str | None = None
+        self, prompt_for_acl: bool = False, acl_config: ACLConfig | dict | str | None = None
     ) -> None:
         """
         Login to Redis using the ACL system. This is the main entry point for the login process, started
@@ -97,9 +127,8 @@ def _bec_service_login(
         Args:
             prompt_for_acl (bool): If True, prompt the user to login using ACL. This is typically only used
                 for user-facing services. Default is False.
-            acl_config (dict or str): The ACL configuration. If a string is provided, it will be treated as a
-                path to the configuration file. If a dictionary is provided, it should contain the username and
-                password for the account.
+            acl_config (ACLConfig, dict, str, optional): The ACL configuration. If a string is provided,
+                it will be treated as a path to the configuration file.
         """
 
         if not self.connector.redis_server_is_running():
@@ -241,40 +270,58 @@ def _local_login(self, selected_account: str) -> str:
         password = getpass(f"Enter the token for {selected_account} (hidden): ")
         return password
 
-    def _config_login_successful(self, prompt_for_acl: bool, acl_config: dict | str) -> bool:
+    def _config_login_successful(
+        self, prompt_for_acl: bool, acl_config: ACLConfig | dict | str | None
+    ) -> bool:
         """
         Login to Redis using the configuration file.
 
         Args:
             prompt_for_acl (bool): If True, prompt the user to login using ACL. Default is False.
-            acl_config (dict or str): The ACL configuration. If a string is provided, it will be treated as a path to the configuration file.
+            acl_config (ACLConfig, dict, str, optional): The ACL configuration. If a string is provided,
+                it will be treated as a path to the configuration file.
 
         Returns:
             bool: True if the login was successful, False otherwise.
         """
 
+        if acl_config is None:
+            return False
+
         if isinstance(acl_config, str):
-            if os.path.exists(acl_config) and not prompt_for_acl:
-                # Load the account information from the .env file
-                # This is relevant for the BEC services that are not launched by the user
-                # but are auto-deployed.
-                account = dotenv_values(acl_config)
-                user = account.get("REDIS_USER")
-                password = account.get("REDIS_PASSWORD")
-                if self._check_redis_auth(user, password):
-                    return True
+            env_file = acl_config
+            username = None
+            password = None
         elif isinstance(acl_config, dict):
-            username = acl_config.get("username")
+            env_file = acl_config.get("env_file")
+            username = acl_config.get("username") or acl_config.get("user")
             password = acl_config.get("password")
-            if self._check_redis_auth(username, password):
-                return True
-            if not password and prompt_for_acl:
-                self._user_service_login(username=username)
-                return True
+        elif isinstance(acl_config, ACLConfig):
+            env_file = acl_config.env_file
+            username = acl_config.user
+            password = acl_config.password
         else:
             raise ValueError(
-                "Invalid value for 'acl' in the service config. Must be a dict or a path to a .env file."
+                "Invalid value for 'acl' in the service config. Must be an ACLConfig, dict, "
+                "or a path to a .env file."
             )
+
+        if env_file and not prompt_for_acl and os.path.exists(env_file):
+            # Load the account information from the .env file. This is relevant for BEC services
+            # that are not launched by the user but are auto-deployed.
+            account = dotenv_values(env_file)
+            user = account.get("REDIS_USER")
+            env_password = account.get("REDIS_PASSWORD")
+            if self._check_redis_auth(user, env_password):
+                return True
+
+        if (username or password) and self._check_redis_auth(username, password):
+            return True
+
+        if prompt_for_acl and username:
+            self._user_service_login(username=username)
+            return True
+
         return False
 
     def _default_user_login_successful(self, full_access: bool) -> bool:

bec_lib/bec_lib/bec_service.py

@@ -6,19 +6,15 @@
 
 import argparse
 import getpass
-import os
 import socket
 import sys
 import threading
 import time
 import uuid
 from dataclasses import asdict, dataclass
-from importlib.metadata import PackageNotFoundError
-from importlib.metadata import version as importlib_version
 from typing import TYPE_CHECKING, Any, Literal
 
 import psutil
-import tomli
 from rich.console import Console
 from rich.table import Table
 
@@ -111,7 +107,7 @@ def parse_cmdline_args(parser=None, config_name: Literal["client", "server"] | s
     config_file = args.config
     cli_args = vars(args)
     user = cli_args.pop("user")
-    acl_config = {"username": user} if user else {}
+    acl_config = {"user": user} if user else None
     redis_url = args.bec_server
 
     if redis_url and config_file:
@@ -134,18 +130,21 @@ def parse_cmdline_args(parser=None, config_name: Literal["client", "server"] | s
             except ValueError:
                 raise ValueError(f"Invalid port number in Redis URL: {comps[1]}")
 
-        service_config = ServiceConfig(
-            redis=redis_data, cmdline_args=cli_args, acl=acl_config, config_name=config_name
-        )
+        kwargs = {"redis": redis_data, "cmdline_args": cli_args, "config_name": config_name}
+        if acl_config:
+            kwargs["acl"] = acl_config
+        service_config = ServiceConfig(**kwargs)
     elif config_file:
-        service_config = ServiceConfig(
-            config_file, cmdline_args=cli_args, acl=acl_config, config_name=config_name
-        )
+        kwargs = {"cmdline_args": cli_args, "config_name": config_name}
+        if acl_config:
+            kwargs["acl"] = acl_config
+        service_config = ServiceConfig(config_file, **kwargs)
     else:
         # If no config file or Redis URL is provided, use the default ServiceConfig
-        service_config = ServiceConfig(
-            cmdline_args=cli_args, acl=acl_config, config_name=config_name
-        )
+        kwargs = {"cmdline_args": cli_args, "config_name": config_name}
+        if acl_config:
+            kwargs["acl"] = acl_config
+        service_config = ServiceConfig(**kwargs)
 
     return args, extra_args, service_config
 
@@ -173,7 +172,7 @@ def __init__(
             else connector
         )
         self.acl = BECAccess(self.connector)
-        self.acl._bec_service_login(prompt_for_acl, self._service_config.config.get("acl"))
+        self.acl._bec_service_login(prompt_for_acl, self._service_config.model.acl)
 
         self._unique_service = unique_service
         self.wait_for_server = wait_for_server

bec_lib/bec_lib/endpoints.py

@@ -1738,18 +1738,18 @@ def gui_registry_state(gui_id: str):
         )
 
     @staticmethod
-    def gui_acl(gui_id: str):
+    def acl_session(session_id: str):
         """
-        Endpoint for exchanging GUI ACL information. This endpoint is used by the CLI or GUI to exchange
+        Endpoint for exchanging ACL session information. This endpoint is used by the CLI or GUI to exchange
         updates on the required ACL user. It uses a messages.CredentialsMessage message.
 
         Args:
-            gui_id (str): GUI ID.
+            session_id (str): Session ID.
         Returns:
-            EndpointInfo: Endpoint for GUI ACL.
+            EndpointInfo: Endpoint for ACL session.
         """
 
-        endpoint = f"{EndpointType.USER.value}/gui/acl/{gui_id}"
+        endpoint = f"{EndpointType.USER.value}/acl/{session_id}"
         return EndpointInfo(
             endpoint=endpoint, message_type=messages.CredentialsMessage, message_op=MessageOp.SEND
         )

bec_lib/bec_lib/service_config.py

@@ -83,6 +83,7 @@ class ACLConfig(BaseModel):
 
     env_file: str = Field(default_factory=lambda: os.path.join(DEFAULT_BASE_PATH, ".bec_acl.env"))
     user: str | None = None
+    password: str | None = None
 
 
 class ProcedureConfig(BaseModel):