Merge pull request #1001 from reef-technologies/escape

Escape Control Characters

b2/_internal/_cli/argcompleters.py

@@ -16,9 +16,11 @@
 
 
 def bucket_name_completer(prefix, parsed_args, **kwargs):
+    from b2sdk.v2 import unprintable_to_hex
+
     from b2._internal._cli.b2api import _get_b2api_for_profile
     api = _get_b2api_for_profile(getattr(parsed_args, 'profile', None))
-    res = [bucket.name for bucket in api.list_buckets(use_cache=True)]
+    res = [unprintable_to_hex(bucket.name) for bucket in api.list_buckets(use_cache=True)]
     return res
 
 
@@ -28,7 +30,7 @@ def file_name_completer(prefix, parsed_args, **kwargs):
 
     To limit delay & cost only lists files returned from by single call to b2_list_file_names
     """
-    from b2sdk.v2 import LIST_FILE_NAMES_MAX_LIMIT
+    from b2sdk.v2 import LIST_FILE_NAMES_MAX_LIMIT, unprintable_to_hex
 
     from b2._internal._cli.b2api import _get_b2api_for_profile
 
@@ -41,7 +43,7 @@ def file_name_completer(prefix, parsed_args, **kwargs):
         fetch_count=LIST_FILE_NAMES_MAX_LIMIT,
     )
     return [
-        folder_name or file_version.file_name
+        unprintable_to_hex(folder_name or file_version.file_name)
         for file_version, folder_name in islice(file_versions, LIST_FILE_NAMES_MAX_LIMIT)
     ]
 
@@ -50,7 +52,7 @@ def b2uri_file_completer(prefix: str, parsed_args, **kwargs):
     """
     Complete B2 URI pointing to a file-like object in a bucket.
     """
-    from b2sdk.v2 import LIST_FILE_NAMES_MAX_LIMIT
+    from b2sdk.v2 import LIST_FILE_NAMES_MAX_LIMIT, unprintable_to_hex
 
     from b2._internal._cli.b2api import _get_b2api_for_profile
     from b2._internal._utils.python_compat import removeprefix
@@ -60,7 +62,10 @@ def b2uri_file_completer(prefix: str, parsed_args, **kwargs):
     if prefix.startswith('b2://'):
         prefix_without_scheme = removeprefix(prefix, 'b2://')
         if '/' not in prefix_without_scheme:
-            return [f"b2://{bucket.name}/" for bucket in api.list_buckets(use_cache=True)]
+            return [
+                f"b2://{unprintable_to_hex(bucket.name)}/"
+                for bucket in api.list_buckets(use_cache=True)
+            ]
 
         b2_uri = parse_b2_uri(prefix)
         bucket = api.get_bucket_by_name(b2_uri.bucket_name)
@@ -72,7 +77,7 @@ def b2uri_file_completer(prefix: str, parsed_args, **kwargs):
             with_wildcard=True,
         )
         return [
-            f"b2://{bucket.name}/{file_version.file_name}"
+            unprintable_to_hex(f"b2://{bucket.name}/{file_version.file_name}")
             for file_version, folder_name in islice(file_versions, LIST_FILE_NAMES_MAX_LIMIT)
             if file_version
         ]

b2/_internal/_cli/autocomplete_install.py

@@ -13,7 +13,6 @@
 import subprocess
 from datetime import datetime
 from pathlib import Path
-from shlex import quote
 from typing import List
 
 import argcomplete
@@ -83,11 +82,11 @@ def get_script_path(self) -> Path:
 
     def program_in_path(self) -> bool:
         """Check if the given program is in PATH."""
-        return _silent_success_run([self.shell_exec, '-c', quote(self.prog)])
+        return _silent_success_run([self.shell_exec, '-c', self.prog])
 
     def is_enabled(self) -> bool:
         """Check if autocompletion is enabled."""
-        return _silent_success_run([self.shell_exec, '-i', '-c', f'complete -p {quote(self.prog)}'])
+        return _silent_success_run([self.shell_exec, '-i', '-c', f'complete -p {self.prog}'])
 
 
 @SHELL_REGISTRY.register('bash')

b2/_internal/_cli/const.py

@@ -20,8 +20,9 @@
 B2_SOURCE_SSE_C_KEY_B64_ENV_VAR = 'B2_SOURCE_SSE_C_KEY_B64'
 
 # Threads defaults
-
 DEFAULT_THREADS = 10
 
 # Constants used in the B2 API
 CREATE_BUCKET_TYPES = ('allPublic', 'allPrivate')
+
+B2_ESCAPE_CONTROL_CHARACTERS = 'B2_ESCAPE_CONTROL_CHARACTERS'

b2/_internal/console_tool.py

@@ -96,10 +96,13 @@
     TqdmProgressListener,
     UploadMode,
     current_time_millis,
+    escape_control_chars,
     get_included_sources,
     make_progress_listener,
     parse_sync_folder,
     points_to_fifo,
+    substitute_control_chars,
+    unprintable_to_hex,
 )
 from b2sdk.v2.exception import (
     B2Error,
@@ -137,6 +140,7 @@
     B2_DESTINATION_SSE_C_KEY_B64_ENV_VAR,
     B2_DESTINATION_SSE_C_KEY_ID_ENV_VAR,
     B2_ENVIRONMENT_ENV_VAR,
+    B2_ESCAPE_CONTROL_CHARACTERS,
     B2_SOURCE_SSE_C_KEY_B64_ENV_VAR,
     B2_USER_AGENT_APPEND_ENV_VAR,
     CREATE_BUCKET_TYPES,
@@ -163,6 +167,22 @@
 # Disable for 1.* behavior.
 VERSION_0_COMPATIBILITY = False
 
+
+class NoControlCharactersStdout:
+    def __init__(self, stdout):
+        self.stdout = stdout
+
+    def __getattr__(self, attr):
+        return getattr(self.stdout, attr)
+
+    def write(self, s):
+        if s:
+            s, cc_present = substitute_control_chars(s)
+            if cc_present:
+                logger.warning('WARNING: Control Characters were detected in the output')
+        self.stdout.write(s)
+
+
 # The name of an executable entry point
 NAME = os.path.basename(sys.argv[0])
 if NAME.endswith('.py'):
@@ -863,6 +883,18 @@ def create_parser(
                 common_parser.add_argument(
                     '-q', '--quiet', action='store_true', default=False, help=argparse.SUPPRESS
                 )
+
+                common_parser.add_argument(
+                    '--escape-control-characters', action='store_true', help=argparse.SUPPRESS
+                )
+                common_parser.add_argument(
+                    '--no-escape-control-characters',
+                    dest='escape_control_characters',
+                    action='store_false',
+                    help=argparse.SUPPRESS
+                )
+
+                common_parser.set_defaults(escape_control_characters=None)
                 parents = [common_parser]
 
             subparsers = parser.add_subparsers(
@@ -902,12 +934,22 @@ def _parse_file_infos(cls, args_info):
 
     def _print_json(self, data) -> None:
         return self._print(
-            json.dumps(data, indent=4, sort_keys=True, cls=B2CliJsonEncoder), enforce_output=True
+            json.dumps(data, indent=4, sort_keys=True, ensure_ascii=True, cls=B2CliJsonEncoder),
+            enforce_output=True
         )
 
-    def _print(self, *args, enforce_output: bool = False, end: str | None = None) -> None:
+    def _print(
+        self,
+        *args,
+        enforce_output: bool = False,
+        end: str | None = None,
+    ) -> None:
         return self._print_standard_descriptor(
-            self.stdout, "stdout", *args, enforce_output=enforce_output, end=end
+            self.stdout,
+            "stdout",
+            *args,
+            enforce_output=enforce_output,
+            end=end,
         )
 
     def _print_stderr(self, *args, end: str | None = None) -> None:
@@ -1011,6 +1053,10 @@ class B2(Command):
     Please note that the above rules may be changed in next versions of b2sdk, and in order to get
     reliable authentication file location you should use ``b2 get-account-info``.
 
+    Control characters escaping is turned on if running under terminal.
+    You can override it by explicitly using `--escape-control-chars`/`--no-escape-control-chars`` option,
+    or by setting `B2_ESCAPE_CONTROL_CHARACTERS` environment variable to either `1` or `0`.
+
     You can suppress command stdout & stderr output by using ``--quiet`` option.
     To supress only progress bar, use ``--noProgress`` option.
 
@@ -2202,17 +2248,23 @@ def _print_file_version(
         file_version: FileVersion,
         folder_name: str | None,
     ) -> None:
-        self._print(folder_name or file_version.file_name)
+        name = folder_name or file_version.file_name
+        if args.escape_control_characters:
+            name = escape_control_chars(name)
+        self._print(name)
 
     def _get_ls_generator(self, args):
-        b2_uri = self.get_b2_uri_from_arg(args)
-        yield from self.api.list_file_versions_by_uri(
-            b2_uri,
-            latest_only=not args.versions,
-            recursive=args.recursive,
-            with_wildcard=args.withWildcard,
-            filters=args.filters,
-        )
+        try:
+            b2_uri = self.get_b2_uri_from_arg(args)
+            yield from self.api.list_file_versions_by_uri(
+                b2_uri,
+                latest_only=not args.versions,
+                recursive=args.recursive,
+                with_wildcard=args.withWildcard,
+                filters=args.filters,
+            )
+        except Exception as err:
+            raise CommandError(unprintable_to_hex(str(err))) from err
 
     def get_b2_uri_from_arg(self, args: argparse.Namespace) -> B2URI:
         raise NotImplementedError
@@ -2273,16 +2325,18 @@ def _print_file_version(
         if not args.long:
             super()._print_file_version(args, file_version, folder_name)
         elif folder_name is not None:
-            self._print(self.format_folder_ls_entry(folder_name, args.replication))
+            self._print(self.format_folder_ls_entry(args, folder_name, args.replication))
         else:
-            self._print(self.format_ls_entry(file_version, args.replication))
+            self._print(self.format_ls_entry(args, file_version, args.replication))
 
-    def format_folder_ls_entry(self, name, replication: bool):
+    def format_folder_ls_entry(self, args, name, replication: bool):
+        if args.escape_control_characters:
+            name = escape_control_chars(name)
         if replication:
             return self.LS_ENTRY_TEMPLATE_REPLICATION % ('-', '-', '-', '-', 0, '-', name)
         return self.LS_ENTRY_TEMPLATE % ('-', '-', '-', '-', 0, name)
 
-    def format_ls_entry(self, file_version: FileVersion, replication: bool):
+    def format_ls_entry(self, args, file_version: FileVersion, replication: bool):
         dt = datetime.datetime.fromtimestamp(
             file_version.upload_timestamp / 1000, datetime.timezone.utc
         )
@@ -2300,7 +2354,10 @@ def format_ls_entry(self, file_version: FileVersion, replication: bool):
         if replication:
             replication_status = file_version.replication_status
             parameters.append(replication_status.value if replication_status else '-')
-        parameters.append(file_version.file_name)
+        name = file_version.file_name
+        if args.escape_control_characters:
+            name = escape_control_chars(name)
+        parameters.append(name)
         return template % tuple(parameters)
 
 
@@ -4020,13 +4077,31 @@ def __init__(self, b2_api: B2Api | None, stdout, stderr):
         self.stdout = stdout
         self.stderr = stderr
 
+    def _get_default_escape_cc_setting(self):
+        escape_cc_env_var = os.environ.get(B2_ESCAPE_CONTROL_CHARACTERS, None)
+        if escape_cc_env_var is not None:
+            if int(escape_cc_env_var) in (0, 1):
+                return int(escape_cc_env_var) == 1
+            else:
+                logger.warning(
+                    "WARNING: invalid value for {B2_ESCAPE_CONTROL_CHARACTERS} environment variable, available options are 0 or 1 - will assume variable is not set"
+                )
+        return self.stdout.isatty()
+
     def run_command(self, argv):
         signal.signal(signal.SIGINT, keyboard_interrupt_handler)
         parser = B2.create_parser(name=argv[0])
         AUTOCOMPLETE.cache_and_autocomplete(parser)
         args = parser.parse_args(argv[1:])
         self._setup_logging(args, argv)
 
+        if args.escape_control_characters is None:
+            args.escape_control_characters = self._get_default_escape_cc_setting()
+
+        if args.escape_control_characters:
+            # in case any control characters slip through escaping, just delete them
+            self.stdout = NoControlCharactersStdout(self.stdout)
+            self.stderr = NoControlCharactersStdout(self.stderr)
         if self.api:
             if (
                 args.profile or getattr(args, 'write_buffer_size', None) or

changelog.d/+escape_control_characters.added.md

@@ -0,0 +1,2 @@
+Added `--escape-control-characters` and `--no-escape-control-characters` flags as well as `B2_ESCAPE_CONTROL_CHARACTERS` env var.
+Control characters escaping is enabled by default if b2 is running in a terminal.

doc/source/conf.py

@@ -42,7 +42,6 @@
 
 B2 = importlib.import_module(f'b2._internal.{LATEST_STABLE_VERSION}.registry').B2
 
-
 # -- General configuration ------------------------------------------------
 
 # If your documentation needs a minimal Sphinx version, state it here.
@@ -53,12 +52,8 @@
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
 extensions = [
-    'sphinx.ext.autodoc',
-    'sphinx.ext.intersphinx',
-    'sphinx.ext.ifconfig',
-    'sphinx.ext.viewcode',
-    'sphinx.ext.coverage',
-    'sphinxarg.ext'
+    'sphinx.ext.autodoc', 'sphinx.ext.intersphinx', 'sphinx.ext.ifconfig', 'sphinx.ext.viewcode',
+    'sphinx.ext.coverage', 'sphinxarg.ext'
 ]
 
 # Add any paths that contain templates here, relative to this directory.
@@ -74,10 +69,10 @@
 master_doc = 'index'
 
 # General information about the project.
-project = u'B2_Command_Line_Tool'
+project = 'B2_Command_Line_Tool'
 
 year = datetime.date.today().strftime("%Y")
-author = u'Backblaze'
+author = 'Backblaze'
 copyright = f'{year}, {author}'
 
 # The version info for the project you're documenting, acts as replacement for
@@ -181,8 +176,8 @@
 #  author, documentclass [howto, manual, or own class]).
 latex_documents = [
     (
-        master_doc, 'B2_Command_Line_Tool.tex', u'B2\\_Command\\_Line\\_Tool Documentation',
-        u'Backblaze', 'manual'
+        master_doc, 'B2_Command_Line_Tool.tex', 'B2\\_Command\\_Line\\_Tool Documentation',
+        'Backblaze', 'manual'
     ),
 ]
 
@@ -191,7 +186,7 @@
 # One entry per manual page. List of tuples
 # (source start file, name, description, authors, manual section).
 man_pages = [
-    (master_doc, 'b2_command_line_tool', u'B2_Command_Line_Tool Documentation', [author], 1)
+    (master_doc, 'b2_command_line_tool', 'B2_Command_Line_Tool Documentation', [author], 1)
 ]
 
 # -- Options for Texinfo output -------------------------------------------
@@ -201,15 +196,14 @@
 #  dir menu entry, description, category)
 texinfo_documents = [
     (
-        master_doc, 'B2_Command_Line_Tool', u'B2_Command_Line_Tool Documentation', author,
+        master_doc, 'B2_Command_Line_Tool', 'B2_Command_Line_Tool Documentation', author,
         'B2_Command_Line_Tool', 'One line description of project.', 'Miscellaneous'
     ),
 ]
 
 # Example configuration for intersphinx: refer to the Python standard library.
 intersphinx_mapping = {'python3': ('https://docs.python.org/3', None)}
 
-
 white_spaces_start = re.compile(r'^\s*')
 
 
@@ -224,7 +218,7 @@ def setup(_):
 
     main_help_path = path.join(path.dirname(__file__), 'main_help.rst')
     if path.exists(main_help_path):
-        with open(main_help_path, 'r') as main_help_file:
+        with open(main_help_path) as main_help_file:
             if main_help_file.read() == main_help_text:
                 return
     with open(main_help_path, 'w') as main_help_file: