Harden file uploads against path traversal.

app/contacts/contact_dns.py

@@ -486,6 +486,9 @@ async def _process_upload_data(self, request_context):
 
     async def _submit_uploaded_file(self, paw, directory, filename, data):
         if paw and filename and directory and data:
+            if not self.file_svc._validate_filename(filename):
+                self.log.warning('Invalid filename rejected: %r', filename)
+                return
             created_dir = os.path.normpath('/' + directory).lstrip('/')
             saveto_dir = await self.file_svc.create_exfil_sub_directory(dir_name=created_dir)
             await self.file_svc.save_file(filename, data, saveto_dir)

app/contacts/contact_ftp.py

@@ -212,6 +212,9 @@ async def get_payload_file(self, payload_dict):
         return await self.file_svc.get_file(payload_dict)
 
     async def submit_uploaded_file(self, paw, filename, data):
+        if not self.file_svc._validate_filename(filename):
+            self.logger.warning('Invalid filename rejected: %r', filename)
+            return
         created_dir = os.path.normpath('/' + paw).lstrip('/')
         saveto_dir = await self.file_svc.create_exfil_sub_directory(dir_name=created_dir)
         await self.file_svc.save_file(filename, data, saveto_dir)

app/contacts/contact_gist.py

@@ -206,9 +206,12 @@ async def _ready_to_export(self, paw, upload_id):
     async def _submit_uploaded_file(self, paw, upload_id):
         upload_info = self.pending_uploads[paw].get(upload_id)
         if upload_info is not None:
+            unique_filename = ''.join([upload_info.filename, '-', upload_id[0:10]])
+            if not self.file_svc._validate_filename(unique_filename):
+                self.log.warning('Invalid filename rejected: %r', unique_filename)
+                return
             created_dir = os.path.normpath('/' + paw).lstrip('/')
             saveto_dir = await self.file_svc.create_exfil_sub_directory(dir_name=created_dir)
-            unique_filename = ''.join([upload_info.filename, '-', upload_id[0:10]])
             await self.file_svc.save_file(unique_filename, upload_info.export_contents(), saveto_dir)
             self.log.debug('Uploaded file %s/%s' % (saveto_dir, upload_info.filename))
 

app/contacts/contact_slack.py

@@ -200,9 +200,12 @@ async def _ready_to_export(self, paw, upload_id):
     async def _submit_uploaded_file(self, paw, upload_id):
         upload_info = self.pending_uploads[paw].get(upload_id)
         if upload_info is not None:
+            unique_filename = ''.join([upload_info.filename, '-', upload_id[0:10]])
+            if not self.file_svc._validate_filename(unique_filename):
+                self.log.warning('Invalid filename rejected: %r', unique_filename)
+                return
             created_dir = os.path.normpath('/' + paw).lstrip('/')
             saveto_dir = await self.file_svc.create_exfil_sub_directory(dir_name=created_dir)
-            unique_filename = ''.join([upload_info.filename, '-', upload_id[0:10]])
             await self.file_svc.save_file(unique_filename, upload_info.export_contents(), saveto_dir)
             self.log.debug('Uploaded file %s/%s' % (saveto_dir, upload_info.filename))
 

app/objects/c_operation.py

@@ -370,7 +370,8 @@ async def run(self, services):
 
     async def write_event_logs_to_disk(self, file_svc, data_svc, output=False):
         event_logs = await self.event_logs(file_svc, data_svc, output=output)
-        event_logs_dir = await file_svc.create_exfil_sub_directory('%s/event_logs' % self.get_config('reports_dir'))
+        event_logs_dir = os.path.join(self.get_config('reports_dir'), 'event_logs')
+        os.makedirs(event_logs_dir, exist_ok=True)
         file_name = 'operation_%s.json' % self.id
         await self._write_logs_to_disk(event_logs, file_name, event_logs_dir, file_svc)
         logging.debug('Wrote event logs for operation %s to disk at %s/%s' % (self.name, event_logs_dir, file_name))

app/service/app_svc.py

@@ -240,7 +240,8 @@ async def _destroy_plugins(self):
 
     async def _write_reports(self):
         file_svc = self.get_service('file_svc')
-        r_dir = await file_svc.create_exfil_sub_directory('%s/reports' % self.get_config('reports_dir'))
+        r_dir = os.path.join(self.get_config('reports_dir'), 'reports')
+        os.makedirs(r_dir, exist_ok=True)
         report = json.dumps(dict(self.get_service('contact_svc').report)).encode()
         await file_svc.save_file('contact_reports', report, r_dir)
         for op in await self.get_service('data_svc').locate('operations'):

app/service/file_svc.py

@@ -76,45 +76,58 @@ async def get_file(self, headers):
     async def save_file(self, filename, payload, target_dir, encrypt=True, encoding=None):
         if encoding:
             payload = await self._decode_contents(payload, encoding)
-        full_path = os.path.join(target_dir, filename)
-        # Containment check — the resolved final path must not escape its parent.
         # save_file has two legitimate call modes:
         #   (1) target_dir + basename — sink mode, used by agent contact handlers
         #       (DNS, FTP, Gist, Slack) that take filename from upload metadata.
         #   (2) ''  + relative path  — path mode, used by ability/source managers
         #       that assemble paths from sanitised IDs.
-        # Either mode is broken by a filename containing '..'. Resolving the final
-        # path and checking it stays under the parent rejects both classes with one
-        # check, without needing callers to know which mode they are using.
-        parent = os.path.realpath(target_dir) if target_dir else os.path.realpath(os.getcwd())
-        final = os.path.realpath(full_path)
-        if final != parent and not final.startswith(parent + os.sep):
-            raise ValueError('save_file: path %r escapes parent %r (filename=%r, target_dir=%r)'
-                             % (final, parent, filename, target_dir))
+        if target_dir:
+            if not self._validate_filename(filename):
+                raise ValueError('save_file: invalid filename %r' % filename)
+            full_path = os.path.join(target_dir, filename)
+            parent = target_dir
+        else:
+            self._validate_relative_save_path(filename)
+            full_path = filename
+            parent = os.getcwd()
+        self._assert_contained(parent, full_path, label='save_file')
         self._save(full_path, payload, encrypt)
 
     async def create_exfil_sub_directory(self, dir_name):
-        path = os.path.join(self.get_config('exfil_dir'), dir_name)
+        if not dir_name or '\x00' in dir_name:
+            raise ValueError('create_exfil_sub_directory: invalid dir_name %r' % dir_name)
+        exfil_dir = self.get_config('exfil_dir')
+        path = os.path.join(exfil_dir, dir_name)
+        self._assert_contained(exfil_dir, path, label='create_exfil_sub_directory')
         if not os.path.exists(path):
             os.makedirs(path)
         return path
 
     async def create_exfil_operation_directory(self, dir_name, agent_name):
+        if not dir_name or '\x00' in dir_name:
+            raise ValueError('create_exfil_operation_directory: invalid dir_name %r' % dir_name)
+        exfil_dir = self.get_config('exfil_dir')
+        self._assert_contained(exfil_dir, dir_name, label='create_exfil_operation_directory')
         op_list = self.data_svc.ram['operations']
         op_list_filtered = [x for x in op_list if x.state not in x.get_finished_states()]
         special_chars = {ord(c): '_' for c in r':<>"/\|?*'}
         agent_opid = [(x.name.translate(special_chars), '_', x.start.strftime("%Y-%m-%d_%H%M%SZ"))
                       for x in op_list_filtered if agent_name in [y.paw for y in x.agents]]
-        path = os.path.join((dir_name), ''.join(agent_opid[0]))
+        if not agent_opid:
+            raise ValueError('create_exfil_operation_directory: no matching operation for agent %r' % agent_name)
+        subdir = ''.join(agent_opid[0])
+        if '\x00' in subdir:
+            raise ValueError('create_exfil_operation_directory: invalid subdirectory %r' % subdir)
+        path = os.path.join(dir_name, subdir)
+        self._assert_contained(dir_name, path, label='create_exfil_operation_directory')
+        self._assert_contained(exfil_dir, path, label='create_exfil_operation_directory')
         if not os.path.exists(path):
             os.makedirs(path)
         return path
 
     @staticmethod
-    def _validate_filename(name):
-        """Validate that a filename contains only safe characters.
-        Returns True if valid, False otherwise.
-        """
+    def _is_safe_path_component(name):
+        """Structural filename checks (no base64). Used for path-mode save_file parts."""
         if not name or len(name) > 255:
             return False
         if '\x00' in name:
@@ -125,6 +138,38 @@ def _validate_filename(name):
             return False
         return True
 
+    @staticmethod
+    def _validate_filename(name):
+        """Validate that a filename contains only safe characters.
+        Returns True if valid, False otherwise.
+        """
+        if not FileSvc._is_safe_path_component(name):
+            return False
+        try:
+            padding = '=' * (-len(name) % 4)
+            decoded_name = base64.b64decode(name + padding, validate=True)
+            if base64.b64encode(decoded_name).decode('ascii').rstrip('=') == name:
+                return False
+        except (binascii.Error, ValueError):
+            pass
+        return True
+
+    @classmethod
+    def _validate_relative_save_path(cls, relpath):
+        if not relpath or os.path.isabs(relpath) or '\x00' in relpath:
+            raise ValueError('save_file: invalid path %r' % relpath)
+        for part in relpath.replace('\\', '/').split('/'):
+            if not part or not cls._is_safe_path_component(part):
+                raise ValueError('save_file: invalid filename component %r in %r' % (part, relpath))
+
+    @staticmethod
+    def _assert_contained(parent, candidate, *, label):
+        parent_real = os.path.realpath(parent)
+        candidate_real = os.path.realpath(candidate)
+        if candidate_real != parent_real and not candidate_real.startswith(parent_real + os.sep):
+            raise ValueError('%s: path %r escapes parent %r' % (label, candidate_real, parent_real))
+        return candidate_real
+
     async def save_multipart_file_upload(self, request, target_dir, encrypt=True):
         try:
             reader = await request.multipart()

tests/services/test_file_svc.py

@@ -61,7 +61,7 @@ def test_save_file_rejects_path_traversal(self, event_loop, file_svc, tmp_path):
             '../../../../tmp/caldera-traversal-canary-3.bin',
         ]
         for evil in traversal_attempts:
-            with pytest.raises(ValueError, match='escapes parent'):
+            with pytest.raises(ValueError, match='invalid filename|escapes parent'):
                 event_loop.run_until_complete(
                     file_svc.save_file(evil, payload, str(tmp_path), encrypt=False)
                 )
@@ -77,6 +77,23 @@ def test_create_exfil_sub_directory(self, event_loop, file_svc):
         assert os.path.isdir(new_dir)
         os.rmdir(new_dir)
 
+    def test_create_exfil_sub_directory_rejects_escape(self, event_loop, file_svc):
+        escape_name = '../caldera-exfil-escape-canary'
+        abs_escape = '/tmp/caldera-exfil-escape-canary'
+        with pytest.raises(ValueError, match='escapes parent'):
+            event_loop.run_until_complete(file_svc.create_exfil_sub_directory(escape_name))
+        with pytest.raises(ValueError, match='escapes parent'):
+            event_loop.run_until_complete(file_svc.create_exfil_sub_directory(abs_escape))
+        exfil_dir = file_svc.get_config('exfil_dir')
+        assert not os.path.exists(os.path.realpath(os.path.join(exfil_dir, escape_name)))
+        assert not os.path.exists(abs_escape)
+
+    def test_create_exfil_operation_directory_rejects_escape(self, event_loop, file_svc):
+        with pytest.raises(ValueError, match='escapes parent'):
+            event_loop.run_until_complete(
+                file_svc.create_exfil_operation_directory('/tmp', 'abcdef')
+            )
+
     def test_read_write_result_file(self, tmpdir, file_svc):
         link_id = '12345'
         output = 'output testing unit'