Multiple security fixes

trashcli/empty/console.py

@@ -1,6 +1,7 @@
 from typing import TextIO
 
 from trashcli.empty.errors import format_error_msg
+from trashcli.lib.sanitize import sanitize_for_output
 
 
 class Console:
@@ -14,10 +15,11 @@ def print_cannot_remove_error(self, path):
         self.print_error("cannot remove %s" % path)
 
     def print_error(self, msg):
-        self.err.write(format_error_msg(self.program_name, msg))
+        self.err.write(format_error_msg(self.program_name,
+                                        sanitize_for_output(msg)))
 
     def print_dry_run(self, path):
-        self.out.write("would remove %s\n" % path)
+        self.out.write("would remove %s\n" % sanitize_for_output(path))
 
     def print_removing(self, path):
-        self.out.write("removing %s\n" % path)
+        self.out.write("removing %s\n" % sanitize_for_output(path))

trashcli/empty/errors.py

@@ -1,10 +1,14 @@
+from trashcli.lib.sanitize import sanitize_for_output
+
+
 class Errors:
     def __init__(self, program_name, err):
         self.program_name = program_name
         self.err = err
 
     def print_error(self, msg):
-        self.err.write(format_error_msg(self.program_name, msg))
+        self.err.write(format_error_msg(self.program_name,
+                                        sanitize_for_output(msg)))
 
 
 def format_error_msg(program_name, msg):

trashcli/lib/sanitize.py

@@ -0,0 +1,21 @@
+from __future__ import absolute_import
+
+
+def sanitize_for_output(s):
+    # Neutralize terminal control sequences in attacker-controlled strings.
+    if s is None:
+        return s
+    out = []
+    for c in s:
+        cp = ord(c)
+        if cp == 0x09:
+            out.append(c)
+        elif cp < 0x20:
+            out.append('^' + chr(cp ^ 0x40))
+        elif cp == 0x7F:
+            out.append('^?')
+        elif 0x80 <= cp <= 0x9F or 0xDC80 <= cp <= 0xDCFF:
+            out.append('\\x%02x' % (cp & 0xFF))
+        else:
+            out.append(c)
+    return ''.join(out)

trashcli/list/list_trash_action.py

@@ -6,10 +6,12 @@
 
 from trashcli.lib.dir_reader import DirReader
 from trashcli.lib.path_of_backup_copy import path_of_backup_copy
+from trashcli.lib.sanitize import sanitize_for_output
 from trashcli.lib.trash_dir_reader import TrashDirReader
 from trashcli.list.extractors import DeletionDateExtractor
 from trashcli.list.extractors import SizeExtractor
-from trashcli.parse_trashinfo.parse_path import parse_path
+from trashcli.parse_trashinfo.parse_original_location import \
+    parse_original_location
 from trashcli.parse_trashinfo.parser_error import ParseError
 from trashcli.trash_dirs_scanner import trash_dir_found
 from trashcli.trash_dirs_scanner import \
@@ -122,13 +124,12 @@ def _print_trashinfo(self,
             yield Error(str(e))
         else:
             try:
-                relative_location = parse_path(contents)
+                original_location = parse_original_location(contents, volume)
             except ParseError:
                 yield Error(self.print_parse_path_error(trashinfo_path))
             else:
                 attribute = extractor.extract_attribute(trashinfo_path,
                                                         contents)
-                original_location = os.path.join(volume, relative_location)
 
                 if show_files:
                     original_file = path_of_backup_copy(trashinfo_path)
@@ -163,8 +164,9 @@ def __init__(self, message):
 
 
 def format_line(attribute, original_location):
-    return "%s %s" % (attribute, original_location)
+    return "%s %s" % (attribute, sanitize_for_output(original_location))
 
 
 def format_line2(attribute, original_location, original_file):
-    return "%s %s -> %s" % (attribute, original_location, original_file)
+    return "%s %s -> %s" % (attribute, sanitize_for_output(original_location),
+                            sanitize_for_output(original_file))

trashcli/parse_trashinfo/parse_original_location.py

@@ -3,8 +3,18 @@
 import os
 
 from trashcli.parse_trashinfo.parse_path import parse_path
+from trashcli.parse_trashinfo.parser_error import ParseError
 
 
 def parse_original_location(contents, volume_path):
+    # Per spec: Path= is absolute only for the home trash (volume=='/');
+    # for volume trashes it must resolve under volume_path.
     path = parse_path(contents)
-    return os.path.join(volume_path, path)
+    resolved = os.path.normpath(os.path.join(volume_path, path))
+    if volume_path != os.path.sep:
+        if os.path.isabs(path):
+            raise ParseError("Path= must be relative for volume trashes")
+        rel = os.path.relpath(resolved, os.path.normpath(volume_path))
+        if rel == '..' or rel.startswith('..' + os.sep):
+            raise ParseError("Path= escapes the volume root")
+    return resolved

trashcli/put/janitor.py

@@ -31,6 +31,13 @@ def log_entries(self, context):  # type: (LogContext) -> str
         return ""
 
 
+class UnableToPersistTrashinfo(NamedTuple('UnableToPersistTrashinfo', [
+    ('error', Exception),
+]), FailureReason):
+    def log_entries(self, context):  # type: (LogContext) -> str
+        return "failed to create trashinfo: %s" % self.error
+
+
 class Janitor:
     def __init__(self,
                  fs,  # type: Fs
@@ -79,7 +86,10 @@ def trash_file_in(self,
             return make_error(trashinfo_data)
 
         persisting_job = self.persister.try_persist(trashinfo_data.value())
-        trashed_file = self.executor.execute(persisting_job, log_data)
+        try:
+            trashed_file = self.executor.execute(persisting_job, log_data)
+        except OSError as e:
+            return Janitor.Result(False, UnableToPersistTrashinfo(e))
         trashed = self.trash_dir.try_trash(trashee.path, trashed_file)
         if isinstance(trashed, Left):
             return make_error(trashed)

trashcli/put/janitor_tools/info_file_persister.py

@@ -50,7 +50,7 @@ def try_persist(self,
                     ):  # type: (...) -> Result
         index = 0
         name_too_long = False
-        while True:
+        while index < 1000:
             suffix = self.suffix.suffix_for_index(index)
             trashinfo_basename = create_trashinfo_basename(data.basename,
                                                            suffix,
@@ -64,6 +64,7 @@ def try_persist(self,
                 self.fs.atomic_write(trashinfo_path, data.content)
                 yield Succeeded(TrashedFile(trashinfo_path),
                                 ".trashinfo created as %s." % trashinfo_path)
+                return
             except OSError as e:
                 if e.errno == errno.ENAMETOOLONG:
                     name_too_long = True

trashcli/put/janitor_tools/security_check.py

@@ -12,6 +12,9 @@ def __init__(self, fs):
     def check_trash_dir_is_secure(self,
                                   candidate,  # type: Candidate
                                   ):  # type: (...) -> Either[None, FailureReason]
+        for sub in (candidate.info_dir(), candidate.files_dir()):
+            if self.fs.lexists(sub) and self.fs.islink(sub):
+                return Left(TrashDirIsNotSecureBecauseSymLink())
         if candidate.check_type == NoCheck:
             return Right(None)
         if candidate.check_type == TopTrashDirCheck:

trashcli/put/jobs.py

@@ -67,8 +67,12 @@ def execute(self,
                 job,  # type: Iterator[JobStatus[R]]
                 log_data,  # type: LogData
                 ):  # type: (...) -> R
+        last = None
         for status in job:
             self.logger.log_multiple(status.logs(), log_data)
             if status.has_succeeded():
                 return status.result()
-        raise ValueError("Should not happen!")
+            last = status
+        raise OSError(
+            "trashinfo creation failed after all retries (last: %s)" %
+            (last.trashinfo_path if last is not None else "?"))

trashcli/put/my_logger.py

@@ -2,6 +2,7 @@
 from typing import List
 
 from trashcli.compat import Protocol
+from trashcli.lib.sanitize import sanitize_for_output
 from trashcli.put.core.logs import Level
 from trashcli.put.core.logs import LogData
 from trashcli.put.core.logs import LogEntry
@@ -27,7 +28,8 @@ def write_message(self,
                       ):
         if is_right_for_level(log_data.verbose, log_entry.level):
             for message in log_entry.resolve_messages():
-                self.stderr.write("%s: %s\n" % (log_data.program_name, message))
+                self.stderr.write("%s: %s\n" % (
+                    log_data.program_name, sanitize_for_output(message)))
 
 
 def is_right_for_level(verbose,  # type: int

trashcli/put/parser.py

@@ -39,6 +39,11 @@ def parse_args(self, argv):  # type: (list) -> Union[ExitWithCode, Trash]
         program_name = os.path.basename(argv[0])
         arg_parser = make_parser(program_name)
         try:
+            bad = _option_shaped_filenames(argv[1:])
+            if bad:
+                arg_parser.error(
+                    "refusing to treat %s as an option "
+                    "(use './%s' or '--' separator)" % (bad[0], bad[0]))
             options = arg_parser.parse_args(argv[1:])
             if len(options.files) <= 0:
                 arg_parser.error("Please specify the files to trash.")
@@ -62,6 +67,22 @@ def ensure_int(code):
     return code
 
 
+def _option_shaped_filenames(args):
+    # Detect argv tokens that look like options AND exist as files on disk:
+    # an attacker-named file reaching argparse can hijack options like --trash-dir.
+    result = []
+    stop = False
+    for a in args:
+        if a == '--':
+            stop = True
+            continue
+        if stop:
+            continue
+        if a.startswith('-') and a != '-' and os.path.lexists(a):
+            result.append(a)
+    return result
+
+
 def make_parser(program_name):
     parser = ArgumentParser(prog=program_name,
                             usage="%(prog)s [OPTION]... FILE...",

trashcli/restore/real_output.py

@@ -2,6 +2,7 @@
 
 import six
 
+from trashcli.lib.sanitize import sanitize_for_output
 from trashcli.restore.output import Output
 from trashcli.restore.output_event import Println, Die, Quit, Exiting, \
     OutputEvent
@@ -17,10 +18,10 @@ def quit(self):
         self.die('')
 
     def printerr(self, msg):
-        print(six.text_type(msg), file=self.stderr)
+        print(sanitize_for_output(six.text_type(msg)), file=self.stderr)
 
     def println(self, line):
-        print(six.text_type(line), file=self.stdout)
+        print(sanitize_for_output(six.text_type(line)), file=self.stdout)
 
     def die(self, error):
         self.printerr(error)

trashcli/restore/restore_asking_the_user.py

@@ -149,7 +149,7 @@ def parse_indexes(user_input,  # type: str
     sequences = []  # type: List[Sequence]
     for index in indexes:
         if "-" in index:
-            first, last = index.split("-", 2)
+            first, last = index.split("-", 1)
             if first == "" or last == "":
                 raise InvalidEntry("open interval: %s" % index)
             split = list(map(parse_int_index, (first, last)))

trashcli/restore/trashed_files.py

@@ -2,6 +2,7 @@
 from typing import NamedTuple
 from typing import Optional
 from typing import Union
+import os
 
 from trashcli.lib.path_of_backup_copy import path_of_backup_copy
 from trashcli.parse_trashinfo.parse_deletion_date import parse_deletion_date
@@ -47,6 +48,10 @@ def all_trashed_files_internal(self,
             if info_file.type == 'non_trashinfo':
                 yield NonTrashinfoFileFound(info_file.path)
             elif info_file.type == 'trashinfo':
+                if not _owned_by_current_user(info_file.path):
+                    yield NonParsableTrashInfo(info_file.path,
+                                               ValueError("not owned by current user"))
+                    continue
                 try:
                     contents = self.file_reader.contents_of(info_file.path)
                     original_location = parse_original_location(contents,
@@ -67,6 +72,17 @@ def all_trashed_files_internal(self,
                                    info_file.type, info_file.path)
 
 
+def _owned_by_current_user(path):
+    # On Windows geteuid is absent; skip the check. Root bypasses so that
+    # `sudo trash-put` followed by an unprivileged restore still works.
+    if not hasattr(os, 'geteuid'):
+        return True
+    try:
+        return os.geteuid() == 0 or os.lstat(path).st_uid == os.geteuid()
+    except OSError:
+        return False
+
+
 class NonTrashinfoFileFound(
     NamedTuple('NonTrashinfoFileFound', [
         ('path', str),