fix: validate upload filename character set in file_svc (#3267)

* fix: validate upload filename character set in file_svc

Reject filenames with path traversal, null bytes, or special characters.

* fix: validate field.filename before os.path.split() to prevent traversal bypass; precompile regex; convert tests to pytest

* fix: flake8 style fixes

* fix: reject '.' as a filename and add test coverage for dot-only names

A single '.' passes the safe-character regex but is not a valid upload
filename. Add an explicit check and a parametrized test case.

* fix: consume rejected multipart part before continue to prevent reader stall

* fix: return 400 Bad Request on invalid upload filename instead of silently skipping

* fix: re-raise HTTPException so HTTPBadRequest propagates; add HTTP-level test for invalid filename

Author: deacon-mp

app/service/file_svc.py

@@ -19,6 +19,7 @@
 from app.utility.base_service import BaseService
 from app.utility.payload_encoder import xor_file, xor_bytes
 
+_SAFE_FILENAME_RE = re.compile(r'^[a-zA-Z0-9._\-]+$')
 FILE_ENCRYPTION_FLAG = '%encrypted%'
 URL_SANITIZATION_REGEX = re.compile(r'^[\w\-\.:%+/]+$')
 ALLOWED_DEFAULT_LDFLAG_REGEX = re.compile(r'^[\w\-\.]+$')
@@ -92,6 +93,21 @@ async def create_exfil_operation_directory(self, dir_name, agent_name):
             os.makedirs(path)
         return path
 
+    @staticmethod
+    def _validate_filename(name):
+        """Validate that a filename contains only safe characters.
+        Returns True if valid, False otherwise.
+        """
+        if not name or len(name) > 255:
+            return False
+        if '\x00' in name:
+            return False
+        if name == '.' or '..' in name or '/' in name or '\\' in name:
+            return False
+        if not _SAFE_FILENAME_RE.fullmatch(name):
+            return False
+        return True
+
     async def save_multipart_file_upload(self, request, target_dir, encrypt=True):
         try:
             reader = await request.multipart()
@@ -100,11 +116,16 @@ async def save_multipart_file_upload(self, request, target_dir, encrypt=True):
                 field = await reader.next()
                 if not field:
                     break
+                if not self._validate_filename(field.filename):
+                    self.log.warning('Invalid filename rejected: %r', field.filename)
+                    raise web.HTTPBadRequest(reason='Invalid filename: disallowed characters or path traversal')
                 _, filename = os.path.split(field.filename)
                 await self.save_file(filename, bytes(await field.read()), target_dir,
                                      encrypt=encrypt, encoding=headers.get('x-file-encoding'))
                 self.log.debug('Uploaded file %s/%s' % (target_dir, filename))
             return web.Response()
+        except web.HTTPException:
+            raise
         except Exception as e:
             self.log.debug('Exception uploading file: %s' % e)
 

tests/security/test_filename_validation.py

@@ -0,0 +1,42 @@
+import pytest
+from app.service.file_svc import FileSvc
+
+
+@pytest.mark.parametrize('name', [
+    'test.ps1',
+    'my-payload_v2.exe',
+    'agent.bin',
+    'a' * 255,
+])
+def test_valid_filenames(name):
+    assert FileSvc._validate_filename(name)
+
+
+@pytest.mark.parametrize('name', [
+    '../../../etc/passwd',
+    '..\\windows\\system32',
+    'test\x00.ps1',
+    'test file.ps1',
+    'test;rm -rf /.ps1',
+    '<script>.js',
+    '',
+    'a' * 256,
+    '.',
+])
+def test_invalid_filenames(name):
+    assert not FileSvc._validate_filename(name)
+
+
+def test_path_traversal_not_bypassed_by_split():
+    """Validate that traversal sequences in the full field.filename are rejected
+    BEFORE os.path.split() strips them — the critical fix for the path traversal bypass.
+    os.path.split('../../../etc/passwd') returns ('../../..', 'passwd');
+    validating only 'passwd' would incorrectly pass."""
+    traversal_filenames = [
+        '../../../etc/passwd',
+        '../../secret.yml',
+        'uploads/../../../etc/shadow',
+    ]
+    for raw_name in traversal_filenames:
+        assert not FileSvc._validate_filename(raw_name), \
+            f'Expected {raw_name!r} to be rejected by _validate_filename (traversal bypass risk)'

tests/services/test_file_svc.py

@@ -4,9 +4,10 @@
 import pytest
 import yaml
 
+from aiohttp import web
 from base64 import b64encode
 from asyncio import Future
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock
 
 from app.data_encoders.base64_basic import Base64Encoder
 from app.data_encoders.plain_text import PlainTextEncoder
@@ -175,6 +176,29 @@ def test_upload_file(self, event_loop, file_svc):
         os.remove(uploaded_file_path)
         os.rmdir(upload_dir)
 
+    def test_multipart_upload_rejects_invalid_filename(self, event_loop, file_svc, tmp_path):
+        """Regression test for #3267: save_multipart_file_upload must raise HTTPBadRequest
+        when a field filename fails _validate_filename (e.g. path traversal attempt '../evil.txt').
+        """
+        # Build a fake multipart field with a path-traversal filename
+        fake_field = MagicMock()
+        fake_field.filename = '../evil.txt'
+        fake_field.read = AsyncMock(return_value=b'evil content')
+
+        # Build a fake multipart reader whose next() returns the bad field then None
+        fake_reader = MagicMock()
+        fake_reader.next = AsyncMock(side_effect=[fake_field, None])
+
+        # Build a fake request whose multipart() returns the reader
+        fake_request = MagicMock()
+        fake_request.multipart = AsyncMock(return_value=fake_reader)
+        fake_request.headers = {}
+
+        with pytest.raises(web.HTTPBadRequest):
+            event_loop.run_until_complete(
+                file_svc.save_multipart_file_upload(fake_request, str(tmp_path))
+            )
+
     def test_encrypt_upload(self, event_loop, file_svc):
         upload_dir = event_loop.run_until_complete(file_svc.create_exfil_sub_directory('test-encrypted-upload'))
         upload_filename = 'encryptedupload.txt'