Optionally guide authentication before establishing boundary tunnel

mycli/boundary_tunnel.py

@@ -1,13 +1,19 @@
 from __future__ import annotations
 
+from collections.abc import Iterator
+from contextlib import ExitStack, contextmanager
 import datetime
 import json
 import os
 import shlex
 import socket
 import subprocess
+import sys
 import threading
 import time
+from typing import IO
+
+from mycli.compat import WIN
 
 
 class BoundaryTunnelError(RuntimeError):
@@ -20,6 +26,32 @@ def _find_free_local_port() -> int:
         return int(sock.getsockname()[1])
 
 
+@contextmanager
+def _authentication_terminal() -> Iterator[tuple[IO[str], IO[str]]]:
+    if sys.stdin.isatty() and sys.stderr.isatty():
+        yield sys.stdin, sys.stderr
+        return
+
+    with ExitStack() as stack:
+        try:
+            if WIN:
+                terminal_input = stack.enter_context(open('CONIN$', 'r', encoding='utf-8'))
+                terminal_output = stack.enter_context(open('CONOUT$', 'w', encoding='utf-8'))
+            else:
+                terminal_input = terminal_output = stack.enter_context(open('/dev/tty', 'r+', encoding='utf-8'))
+        except OSError as exc:
+            raise BoundaryTunnelError('Unable to open a terminal for Boundary authentication.') from exc
+        yield terminal_input, terminal_output
+
+
+def _prompt_for_authentication(terminal_input: IO[str], terminal_output: IO[str], prompt: str) -> str:
+    print(prompt, file=terminal_output, end='', flush=True)
+    response = terminal_input.readline()
+    if not response:
+        raise BoundaryTunnelError('Unable to read a response from the terminal.')
+    return response.rstrip('\r\n')
+
+
 class BoundaryTunnel:
     def __init__(
         self,
@@ -29,6 +61,8 @@ def __init__(
         address: str | None = None,
         auth_method_id: str | None = None,
         boundary_options: str | None = None,
+        boundary_test_command: str | None = None,
+        boundary_auth_command: str | None = None,
         local_port: int | None = None,
         ready_timeout: float = 30.0,
     ) -> None:
@@ -37,6 +71,8 @@ def __init__(
         self.address = address
         self.auth_method_id = auth_method_id
         self.boundary_options = boundary_options
+        self.boundary_test_command = boundary_test_command
+        self.boundary_auth_command = boundary_auth_command
         self.local_host = '127.0.0.1'
         self.local_port = local_port or _find_free_local_port()
         self.ready_timeout = ready_timeout
@@ -68,6 +104,7 @@ def command(self) -> list[str]:
         return command
 
     def start(self, *, show_expiration_warning: bool = True) -> None:
+        self._authenticate_if_needed()
         self._thread = threading.Thread(target=self._run, name='mycli-boundary-tunnel', daemon=True)
         self._thread.start()
         deadline = time.monotonic() + self.ready_timeout
@@ -78,7 +115,7 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
             time.sleep(0.05)
         else:
             self.close()
-            raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process to start.')
+            raise BoundaryTunnelError('Timed out waiting for tunnel process to start.')
 
         while time.monotonic() < deadline:
             self._raise_if_failed()
@@ -87,18 +124,18 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
             time.sleep(0.05)
         else:
             self.close()
-            raise BoundaryTunnelError('Timed out waiting for Boundary tunnel process output.')
+            raise BoundaryTunnelError('Timed out waiting for tunnel process output.')
 
         connection_details = json.loads(self.stdout)
 
         if 'status_code' in connection_details:
-            raise BoundaryTunnelError(f'Boundary tunnel CLI raised status code {connection_details["status_code"]}.')
+            raise BoundaryTunnelError(f'Tunnel CLI raised status code {connection_details["status_code"]}.')
 
         try:
             self.username = connection_details['credentials'][0]['secret']['decoded']['username']
             self.password = connection_details['credentials'][0]['secret']['decoded']['password']
         except (IndexError, KeyError):
-            raise BoundaryTunnelError('Boundary tunnel CLI did not return credentials.') from None
+            raise BoundaryTunnelError('Tunnel CLI did not return credentials.') from None
 
         expiry_raw = connection_details['expiration']
         expiry_utc = datetime.datetime.strptime(expiry_raw, '%Y-%m-%dT%H:%M:%S.%f%z')
@@ -112,15 +149,15 @@ def start(self, *, show_expiration_warning: bool = True) -> None:
                 return
             time.sleep(0.05)
         self.close()
-        raise BoundaryTunnelError('Timed out waiting for Boundary tunnel to become ready.')
+        raise BoundaryTunnelError('Timed out waiting for tunnel to become ready.')
 
     def _raise_if_failed(self) -> None:
         if not self._failed.is_set():
             return
         self.close()
         if self._startup_error is not None:
-            raise BoundaryTunnelError(f'Unable to start Boundary tunnel process: {self._startup_error}') from self._startup_error
-        raise BoundaryTunnelError('Boundary tunnel process exited before it was ready.')
+            raise BoundaryTunnelError(f'Unable to start tunnel process: {self._startup_error}') from self._startup_error
+        raise BoundaryTunnelError('Tunnel process exited before it was ready.')
 
     def _read_stdout(self) -> None:
         process = self.process
@@ -129,12 +166,77 @@ def _read_stdout(self) -> None:
         self.stdout = process.stdout.readline().decode('utf-8')
 
     def _environment(self) -> dict[str, str] | None:
-        if not self.auth_method_id:
-            return None
         environment = os.environ.copy()
-        environment['BOUNDARY_AUTH_METHOD_ID'] = self.auth_method_id
+        if self.auth_method_id:
+            environment['BOUNDARY_AUTH_METHOD_ID'] = self.auth_method_id
+        if self.address:
+            environment['BOUNDARY_ADDR'] = self.address
         return environment
 
+    def _authenticate_if_needed(self) -> None:
+        if not self.boundary_test_command or not self.boundary_auth_command:
+            return
+
+        test_command = self._parse_authentication_command(self.boundary_test_command, 'test')
+        try:
+            completed_process = subprocess.run(
+                test_command,
+                check=False,
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                env=self._environment(),
+            )
+        except OSError as exc:
+            raise BoundaryTunnelError(f'Unable to run test command: {exc}') from exc
+        if completed_process.returncode == 0:
+            return
+
+        with _authentication_terminal() as (terminal_input, terminal_output):
+            yn = _prompt_for_authentication(
+                terminal_input,
+                terminal_output,
+                'Authenticate with Boundary before connecting? [Yn] ',
+            ).lower()
+            if yn not in ('y', ''):
+                raise BoundaryTunnelError('Not authenticated.')
+
+            auth_command = self._parse_authentication_command(self.boundary_auth_command, 'authentication')
+            try:
+                completed_process = subprocess.run(
+                    auth_command,
+                    check=False,
+                    stdin=terminal_input,
+                    stdout=terminal_output,
+                    stderr=terminal_output,
+                    env=self._environment(),
+                )
+            except OSError as exc:
+                raise BoundaryTunnelError(f'Unable to run authentication command: {exc}') from exc
+            if completed_process.returncode != 0:
+                raise BoundaryTunnelError(f'Authentication command exited with status {completed_process.returncode}.')
+
+            _prompt_for_authentication(
+                terminal_input,
+                terminal_output,
+                'Press return to continue after authenticating: ',
+            )
+
+    @staticmethod
+    def _parse_authentication_command(command: str, name: str) -> list[str]:
+        try:
+            arguments = shlex.split(command, posix=not WIN)
+        except ValueError as exc:
+            raise BoundaryTunnelError(f'Unable to parse {name} command: {exc}') from exc
+        if WIN:
+            arguments = [
+                argument[1:-1] if len(argument) >= 2 and argument[0] == argument[-1] and argument[0] in ('"', "'") else argument
+                for argument in arguments
+            ]
+        if not arguments:
+            raise BoundaryTunnelError(f'{name} command is empty.')
+        return arguments
+
     def close(self) -> None:
         process = self.process
         if process is not None and process.poll() is None:

mycli/client_connection.py

@@ -136,13 +136,17 @@ def connect(
             boundary_address = self.config.get('boundary_beta', {}).get('address') or None
             boundary_auth_method_id = self.config.get('boundary_beta', {}).get('auth_method_id') or None
             boundary_options = self.config.get('boundary_beta', {}).get('boundary_options') or None
+            boundary_test_command = self.config.get('boundary_beta', {}).get('boundary_test_command') or None
+            boundary_auth_command = self.config.get('boundary_beta', {}).get('boundary_auth_command') or None
             try:
                 self.boundary_tunnel = BoundaryTunnel(
                     target_id=boundary_target_id,
                     boundary_executable=boundary_executable,
                     address=boundary_address,
                     auth_method_id=boundary_auth_method_id,
                     boundary_options=boundary_options,
+                    boundary_test_command=boundary_test_command,
+                    boundary_auth_command=boundary_auth_command,
                 )
                 self.boundary_tunnel.start(show_expiration_warning=self.verbosity >= 0)
                 if self.verbosity >= 0:

mycli/myclirc

@@ -435,6 +435,18 @@ auth_method_id =
 # Additional options passed to the "boundary connect" command.
 boundary_options =
 
+# Command used to test whether Boundary authentication is valid.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_auth_command must also be set.
+# Example: boundary targets read -id ttcp_zzzzzzzzzz
+boundary_test_command =
+
+# Command used to authenticate when boundary_test_command exits nonzero.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_test_command must also be set.
+# Example: boundary authenticate oidc
+boundary_auth_command =
+
 # Custom colors for the completion menu, toolbar, etc, with actual support
 # depending on the terminal, and the property being set.
 # Colors: #ffffff, bg:#ffffff, border:#ffffff.

test/myclirc

@@ -435,6 +435,18 @@ auth_method_id =
 # Additional options passed to the "boundary connect" command.
 boundary_options =
 
+# Command used to test whether Boundary authentication is valid.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_auth_command must also be set.
+# Example: boundary targets read -id ttcp_zzzzzzzzzz
+boundary_test_command =
+
+# Command used to authenticate when boundary_test_command exits nonzero.
+# If empty, establishing the tunnel will fail when not authenticated.
+# If set, boundary_test_command must also be set.
+# Example: boundary authenticate oidc
+boundary_auth_command =
+
 # Custom colors for the completion menu, toolbar, etc, with actual support
 # depending on the terminal, and the property being set.
 # Colors: #ffffff, bg:#ffffff, border:#ffffff.