feat: Proxy authentication mode with mock token

src/fabric_cli/commands/auth/fab_auth.py

@@ -14,6 +14,17 @@
 
 
 def init(args: Namespace) -> Any:
+    if FabAuth().is_proxy_auth_mode():
+        fab_ui.print_output_error(
+            FabricCLIError(
+                ErrorMessages.Auth.login_not_available_in_proxy_mode(),
+                fab_constant.ERROR_AUTHENTICATION_FAILED,
+            ),
+            command=args.command,
+            output_format_type=args.output_format,
+        )
+        return
+
     auth_options = [
         "Interactive with a web browser",
         "Azure CLI (existing 'az login' session)",
@@ -209,6 +220,17 @@ def init(args: Namespace) -> Any:
 
 
 def logout(args: Namespace) -> None:
+    if FabAuth().is_proxy_auth_mode():
+        fab_ui.print_output_error(
+            FabricCLIError(
+                ErrorMessages.Auth.logout_not_available_in_proxy_mode(),
+                fab_constant.ERROR_AUTHENTICATION_FAILED,
+            ),
+            command=args.command,
+            output_format_type=args.output_format,
+        )
+        return
+
     FabAuth().logout()
 
     # Clear cache and context including current and stale context files
@@ -220,6 +242,10 @@ def logout(args: Namespace) -> None:
 
 def status(args: Namespace) -> None:
     auth = FabAuth()
+    if auth.is_proxy_auth_mode():
+        fab_ui.print_output_format(args, data="proxy authentication mode")
+        return
+
     identity_type = auth.get_identity_type()
     tenant_id = auth.get_tenant_id()
 

src/fabric_cli/core/fab_auth.py

@@ -28,6 +28,17 @@
 from fabric_cli.errors import ErrorMessages
 from fabric_cli.utils import fab_ui as utils_ui
 
+_PROXY_AUTH_ENVIRONMENT_VARIABLE = "FAB_PROXY_AUTH_ENABLED"
+_PROXY_AUTH_PLACEHOLDER_TOKEN = "mockToken"
+_PROXY_AUTH_EXPIRES_ON = 9999999999
+
+
+def _is_proxy_auth_placeholder(token: Any) -> bool:
+    return token in (
+        _PROXY_AUTH_PLACEHOLDER_TOKEN,
+        _PROXY_AUTH_PLACEHOLDER_TOKEN.encode(),
+    )
+
 
 def singleton(class_):
     instances = {}
@@ -128,6 +139,9 @@ def _validate_environment_variables(self):
                     )
 
     def _load_env(self):
+        if self.is_proxy_auth_mode():
+            return
+
         # Validate the environment variables
         self._validate_environment_variables()
 
@@ -328,6 +342,14 @@ def get_tenant_id(self):
     def get_identity_type(self):
         return self._get_auth_property(con.IDENTITY_TYPE)
 
+    @staticmethod
+    def is_proxy_auth_mode() -> bool:
+        """Return whether proxy authentication mode is enabled."""
+        return os.environ.get(_PROXY_AUTH_ENVIRONMENT_VARIABLE, "").lower() in (
+            "true",
+            "1",
+        )
+
     def set_access_mode(self, mode, tenant_id=None):
         if mode not in con.AUTH_KEYS[con.IDENTITY_TYPE]:
             raise FabricCLIError(
@@ -525,6 +547,12 @@ def acquire_token(self, scope: list[str], interactive_renew=True) -> dict:
         from fabric_cli.utils.fab_secure_io import restrict_existing_file
 
         try:
+            if self.is_proxy_auth_mode():
+                return {
+                    "access_token": _PROXY_AUTH_PLACEHOLDER_TOKEN,
+                    "expires_on": _PROXY_AUTH_EXPIRES_ON,
+                }
+
             token = None
             env_var_token = self._get_access_token_from_env_vars_if_exist(scope)
             identity_type = self.get_identity_type()
@@ -677,6 +705,9 @@ def _fetch_public_key_from_aad(self, token):
         return key
 
     def _decode_jwt_token(self, token, expected_audience=None):
+        if self.is_proxy_auth_mode() and _is_proxy_auth_placeholder(token):
+            return {}
+
         decode_options = {"verify_aud": expected_audience is not None}
         # Try using the cached public key if available
         if self.aad_public_key is not None:

src/fabric_cli/errors/auth.py

@@ -120,6 +120,20 @@ def cert_read_failed(error: str) -> str:
     def only_supported_with_user_authentication() -> str:
         return "This operation is only supported with user authentication"
 
+    @staticmethod
+    def login_not_available_in_proxy_mode() -> str:
+        return (
+            "Authentication login is not available in proxy authentication mode. "
+            "Unset FAB_PROXY_AUTH_ENABLED to manage CLI authentication"
+        )
+
+    @staticmethod
+    def logout_not_available_in_proxy_mode() -> str:
+        return (
+            "Authentication logout is not available in proxy authentication mode. "
+            "Unset FAB_PROXY_AUTH_ENABLED to manage CLI authentication"
+        )
+
     @staticmethod
     def azure_cli_not_available() -> str:
         return (

tests/conftest.py

@@ -101,6 +101,7 @@ def azure_cli_auth_fixture(monkeypatch, tmp_path):
         "fabric_cli.core.fab_state_config.config_location", lambda: str(tmp_path)
     )
     for variable in (
+        "FAB_PROXY_AUTH_ENABLED",
         "FAB_TOKEN",
         "FAB_TOKEN_ONELAKE",
         "FAB_TOKEN_AZURE",

tests/test_commands/test_auth.py

@@ -912,6 +912,68 @@ def test_auth_logout(
 
         mock_print_done.assert_called_once()
 
+    def test_auth_login_proxy_auth_mode_no_side_effects_success(self, mock_fab_auth):
+        args = prepare_auth_args()
+        auth = mock_fab_auth["instance"]
+        auth.is_proxy_auth_mode.return_value = True
+
+        with (
+            patch(
+                "fabric_cli.commands.auth.fab_auth.utils_mem_store.clear_caches"
+            ) as clear_caches,
+            patch(
+                "fabric_cli.commands.auth.fab_auth.fab_ui.prompt_select_item"
+            ) as prompt_select_item,
+            patch(
+                "fabric_cli.commands.auth.fab_auth.fab_ui.print_output_error"
+            ) as print_error,
+        ):
+            result = fab_auth.init(args)
+
+        assert result is None
+        error = print_error.call_args.args[0]
+        assert error.status_code == fab_constant.ERROR_AUTHENTICATION_FAILED
+        assert error.message == ErrorMessages.Auth.login_not_available_in_proxy_mode()
+        print_error.assert_called_once_with(
+            error,
+            command=args.command,
+            output_format_type=args.output_format,
+        )
+        assert_fab_auth_not_called(mock_fab_auth)
+        clear_caches.assert_not_called()
+        prompt_select_item.assert_not_called()
+
+    def test_auth_logout_proxy_auth_mode_no_side_effects_success(
+        self, mock_fab_auth, mock_fab_context
+    ):
+        args = argparse.Namespace(command="auth", output_format="text")
+        auth = mock_fab_auth["instance"]
+        auth.is_proxy_auth_mode.return_value = True
+        context = mock_fab_context["instance"]
+
+        with (
+            patch(
+                "fabric_cli.commands.auth.fab_auth.utils_mem_store.clear_caches"
+            ) as clear_caches,
+            patch(
+                "fabric_cli.commands.auth.fab_auth.fab_ui.print_output_error"
+            ) as print_error,
+        ):
+            result = fab_auth.logout(args)
+
+        assert result is None
+        error = print_error.call_args.args[0]
+        assert error.status_code == fab_constant.ERROR_AUTHENTICATION_FAILED
+        assert error.message == ErrorMessages.Auth.logout_not_available_in_proxy_mode()
+        print_error.assert_called_once_with(
+            error,
+            command="auth",
+            output_format_type="text",
+        )
+        auth.logout.assert_not_called()
+        clear_caches.assert_not_called()
+        context.reset_context.assert_not_called()
+
     def test_auth_status(self, mock_fab_auth, capsys):
         # Arrange
         args = argparse.Namespace(
@@ -945,6 +1007,19 @@ def test_auth_status(self, mock_fab_auth, capsys):
         assert "Token Storage: mock************************************" in captured.out
         assert "Token Azure: mock************************************" in captured.out
 
+    def test_auth_status_proxy_auth_mode_success(self, monkeypatch, capsys):
+        monkeypatch.setenv("FAB_PROXY_AUTH_ENABLED", "true")
+        args = argparse.Namespace(
+            command="auth",
+            auth_subcommand="status",
+            output_format="text",
+        )
+        fab_auth.status(args)
+
+        captured = capsys.readouterr()
+        assert captured.out.strip() == "proxy authentication mode"
+        assert captured.err == ""
+
     def test_auth_status_azure_cli_session_available(self, mock_fab_auth, capsys):
         args = argparse.Namespace(
             command="auth",
@@ -1217,6 +1292,7 @@ def mock_fab_auth():
         set_spn=MagicMock(),
         set_managed_identity=MagicMock(),
         logout=MagicMock(),
+        is_proxy_auth_mode=MagicMock(return_value=False),
         # add more methods if needed
     ) as mocks:
         # mocks is a dictionary containing the mock objects for each method

tests/test_core/test_fab_api_client.py

@@ -345,6 +345,53 @@ def __init__(self, status_code, headers=None, text=""):
     mock_sleep.assert_called_once_with(10)
 
 
+@pytest.mark.parametrize("audience", [None, "storage", "azure"])
+def test_do_request_proxy_auth_mode_sends_placeholder_header_success(
+    monkeypatch, audience
+):
+    monkeypatch.setenv("FAB_PROXY_AUTH_ENABLED", "true")
+    monkeypatch.delenv("FAB_TOKEN", raising=False)
+    monkeypatch.delenv("FAB_TOKEN_ONELAKE", raising=False)
+    monkeypatch.delenv("FAB_TOKEN_AZURE", raising=False)
+
+    auth = FabAuth()
+    auth._auth_info = {}
+    monkeypatch.setattr(
+        auth,
+        "_get_app",
+        lambda: pytest.fail("Proxy authentication must not initialize MSAL"),
+    )
+    monkeypatch.setattr(
+        auth,
+        "_decode_jwt_token",
+        lambda token, expected_audience=None: pytest.fail(
+            "Proxy authentication must not decode the placeholder token"
+        ),
+    )
+
+    class DummyResponse:
+        status_code = 200
+        text = "{}"
+        content = b"{}"
+        headers = {}
+
+    dummy_args = Namespace(
+        uri="items",
+        method="get",
+        audience=audience,
+        headers=None,
+        wait=False,
+        raw_response=True,
+        request_params={},
+        json_file=None,
+    )
+
+    with patch("requests.Session.request", return_value=DummyResponse()) as request:
+        do_request(dummy_args)
+
+    assert request.call_args.kwargs["headers"]["Authorization"] == "Bearer mockToken"
+
+
 @pytest.mark.parametrize(
     "host_app_env, host_app_version_env, expected_suffix",
     [