Handle docker persistent login error and keeper get lookup

keepercommander/auth/console_ui.py

@@ -18,6 +18,31 @@ def _stderr(msg=''):
     print(msg, file=sys.stderr)
 
 
+_HEADLESS_AUTH_MSG_SHOWN = False
+
+
+def _is_interactive():
+    try:
+        return bool(sys.stdin) and sys.stdin.isatty()
+    except Exception:
+        return False
+
+
+def _fail_headless_auth(step):
+    """In headless/service mode, persistent login often needs a follow-up prompt
+    (password, SSO, 2FA, device approval) that cannot be answered. Log once and
+    cancel so the caller exits cleanly instead of looping or spamming getpass."""
+    global _HEADLESS_AUTH_MSG_SHOWN
+    if not _HEADLESS_AUTH_MSG_SHOWN:
+        _HEADLESS_AUTH_MSG_SHOWN = True
+        logging.error(
+            'Persistent login is not working in this non-interactive environment '
+            '(possibly due to an IP/location change). '
+            'Re-run Commander/Docker setup from this network, then restart the service.'
+        )
+    step.cancel()
+
+
 class ConsoleLoginUi(login_steps.LoginUi):
     def __init__(self):
         self._show_device_approval_help = True
@@ -28,6 +53,9 @@ def __init__(self):
         self._failed_password_attempt = 0
 
     def on_device_approval(self, step):
+        if not _is_interactive():
+            _fail_headless_auth(step)
+            return
         if self._show_device_approval_help:
             _stderr(f"\n{Fore.YELLOW}Device Approval Required{Fore.RESET}\n")
             _stderr(f"{Fore.CYAN}Select an approval method:{Fore.RESET}")
@@ -123,6 +151,9 @@ def two_factor_channel_to_desc(channel):   # type: (login_steps.TwoFactorChannel
             return 'Backup Codes'
 
     def on_two_factor(self, step):
+        if not _is_interactive():
+            _fail_headless_auth(step)
+            return
         channels = step.get_channels()
 
         if self._show_two_factor_help:
@@ -273,6 +304,9 @@ def on_two_factor(self, step):
                 logging.warning(f'{Fore.YELLOW}Invalid 2FA code. Please try again.{Fore.RESET}')
 
     def on_password(self, step):
+        if not _is_interactive():
+            _fail_headless_auth(step)
+            return
         if self._show_password_help:
             _stderr(f'{Fore.CYAN}Enter master password for {Fore.WHITE}{step.username}{Fore.RESET}')
 
@@ -293,6 +327,9 @@ def on_password(self, step):
                 step.cancel()
 
     def on_sso_redirect(self, step):
+        if not _is_interactive():
+            _fail_headless_auth(step)
+            return
         try:
             wb = webbrowser.get()
             wrappers = set('xdg-open|gvfs-open|gnome-open|x-www-browser|www-browser'.split('|'))
@@ -360,6 +397,9 @@ def on_sso_redirect(self, step):
                 break
 
     def on_sso_data_key(self, step):
+        if not _is_interactive():
+            _fail_headless_auth(step)
+            return
         if self._show_sso_data_key_help:
             _stderr(f'\n{Fore.YELLOW}Device Approval Required for SSO{Fore.RESET}\n')
             _stderr(f'{Fore.CYAN}Select an approval method:{Fore.RESET}')

keepercommander/commands/helpers/record.py

@@ -1,9 +1,26 @@
+import re
 from typing import Set, Optional
 
 from ... import api
+from ...error import CommandError
 from ...params import KeeperParams
 from ...subfolder import try_resolve_path
 
+# Block shell chaining markers in `get` lookup tokens.
+_GET_LOOKUP_CONTROL_CHARS_RE = re.compile(r'[\r\n\x00]')
+_GET_LOOKUP_SHELL_METACHAR_RE = re.compile(r'[;|]')
+_GET_LOOKUP_CHAIN_RE = re.compile(r'&&')
+
+
+def raise_if_unsafe_get_lookup_token(token, command='get'):
+    # type: (str, str) -> None
+    if not token:
+        raise CommandError(command, 'Invalid record identifier: forbidden characters')
+    if (_GET_LOOKUP_CONTROL_CHARS_RE.search(token)
+            or _GET_LOOKUP_SHELL_METACHAR_RE.search(token)
+            or _GET_LOOKUP_CHAIN_RE.search(token)):
+        raise CommandError(command, 'Invalid record identifier: forbidden characters')
+
 
 # Get record UID(s) given one of its identifiers: name (if current folder contains the record), path, or UID
 def get_record_uids(params, name):  # type: (KeeperParams, str) -> Set[Optional[str]]

keepercommander/commands/record.py

@@ -20,6 +20,7 @@
 import re
 from functools import reduce
 from typing import Dict, Any, List, Optional, Iterable, Tuple, Set
+from .helpers.record import raise_if_unsafe_get_lookup_token
 
 from colorama import Fore, Back, Style
 
@@ -294,6 +295,8 @@ def execute(self, params, **kwargs):
         if not uid:
             raise CommandError('get', 'UID parameter is required')
 
+        raise_if_unsafe_get_lookup_token(uid)
+
         fmt = kwargs.get('format') or 'detail'
 
         # First try to interpret as UID

unit-tests/test_command_record.py

@@ -248,6 +248,14 @@ def test_get_invalid_uid(self):
         with self.assertRaises(CommandError):
             cmd.execute(params, uid='invalid')
 
+    def test_get_rejects_shell_metacharacters_in_lookup_token(self):
+        params = get_synced_params()
+        cmd = record.RecordGetUidCommand()
+
+        with self.assertRaises(CommandError) as context:
+            cmd.execute(params, uid='x;cd $HOME && id > pwned_keeper_rce.txt;#"unclosed')
+        self.assertIn('forbidden characters', context.exception.message)
+
     def test_append_notes_command(self):
         params = get_synced_params()
         cmd = record_edit.RecordAppendNotesCommand()