diff --git a/keepercommander/auth/console_ui.py b/keepercommander/auth/console_ui.py index 0a60b077b..eca6bf3b4 100644 --- a/keepercommander/auth/console_ui.py +++ b/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}') diff --git a/keepercommander/commands/helpers/record.py b/keepercommander/commands/helpers/record.py index 7769f5f7e..a2f97976f 100644 --- a/keepercommander/commands/helpers/record.py +++ b/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]] diff --git a/keepercommander/commands/record.py b/keepercommander/commands/record.py index 0c8d152de..99355974c 100644 --- a/keepercommander/commands/record.py +++ b/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 diff --git a/unit-tests/test_command_record.py b/unit-tests/test_command_record.py index fc61833c9..ebfeb2f8f 100644 --- a/unit-tests/test_command_record.py +++ b/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()