From 7eac6971caeed0ca35bd012dc6cb72c9ccb67cc3 Mon Sep 17 00:00:00 2001 From: webmixgamer Date: Sat, 17 Jan 2026 17:55:04 +0000 Subject: [PATCH] security: implement safe tar extraction with symlink/hardlink validation Add comprehensive archive validation for local agent deployment to prevent path traversal attacks via symlinks and hardlinks. Changes: - Add _is_path_within() helper for path containment checks using resolve() - Add _validate_tar_member() to validate each archive member: - Rejects absolute paths and path traversal (../) - Rejects symlinks/hardlinks pointing outside extraction directory - Rejects device files (chr/blk) and FIFOs - Allows internal symlinks/hardlinks that resolve within temp_dir - Add _safe_extract_tar() wrapper that validates all members before extraction - Replace simple startswith/'..' check with full validation - Add unit tests for validation logic (tests/test_archive_security.py) Addresses H-02 finding from security scan. --- src/backend/services/agent_service/deploy.py | 171 ++++++++-- tests/test_archive_security.py | 319 +++++++++++++++++++ 2 files changed, 468 insertions(+), 22 deletions(-) create mode 100644 tests/test_archive_security.py diff --git a/src/backend/services/agent_service/deploy.py b/src/backend/services/agent_service/deploy.py index 28057df50..569f23879 100644 --- a/src/backend/services/agent_service/deploy.py +++ b/src/backend/services/agent_service/deploy.py @@ -36,6 +36,150 @@ MAX_FILES = 1000 +# ============================================================================= +# Safe Tar Extraction Utilities +# ============================================================================= + +def _is_path_within(base: Path, target: Path) -> bool: + """ + Check if target path is within base directory. + + Uses Path.resolve() to handle symlinks and normalize paths. + Returns True if target is inside base (or is base itself). + """ + try: + # resolve() normalizes the path and resolves symlinks + # We use strict=False because target may not exist yet during extraction + base_resolved = base.resolve() + target_resolved = target.resolve() + + # Check if target starts with base path + return str(target_resolved).startswith(str(base_resolved) + "/") or \ + target_resolved == base_resolved + except (OSError, ValueError): + return False + + +def _validate_tar_member(member: tarfile.TarInfo, base_dir: Path) -> tuple[bool, str]: + """ + Validate a tar archive member for safe extraction. + + Checks: + - Destination path stays within base_dir + - No absolute paths + - Symlinks/hardlinks only point within base_dir + - No special file types (devices, FIFOs) + + Args: + member: The tar archive member to validate + base_dir: The base directory for extraction + + Returns: + Tuple of (is_valid, error_message). If valid, error_message is empty. + """ + member_name = member.name + + # Reject absolute paths + if member_name.startswith('/'): + return False, f"Absolute path not allowed: {member_name}" + + # Reject path traversal in member name + if '..' in member_name.split('/'): + return False, f"Path traversal not allowed: {member_name}" + + # Calculate the destination path + dest_path = base_dir / member_name + + # Verify destination stays within base_dir + if not _is_path_within(base_dir, dest_path): + return False, f"Path escapes extraction directory: {member_name}" + + # Reject special file types (devices, FIFOs) + if member.ischr() or member.isblk(): + return False, f"Device files not allowed: {member_name}" + if member.isfifo(): + return False, f"FIFO files not allowed: {member_name}" + + # Validate symlinks + if member.issym(): + linkname = member.linkname + + # Reject absolute symlink targets + if linkname.startswith('/'): + return False, f"Absolute symlink target not allowed: {member_name} -> {linkname}" + + # Calculate where the symlink would point + # Symlink is relative to the directory containing it + link_dir = dest_path.parent + link_target = (link_dir / linkname).resolve() + + # Verify symlink target stays within base_dir + if not _is_path_within(base_dir, link_target): + return False, f"Symlink escapes extraction directory: {member_name} -> {linkname}" + + # Validate hardlinks + if member.islnk(): + linkname = member.linkname + + # Reject absolute hardlink targets + if linkname.startswith('/'): + return False, f"Absolute hardlink target not allowed: {member_name} -> {linkname}" + + # Hardlink target is relative to archive root (base_dir) + link_target = base_dir / linkname + + # Verify hardlink target stays within base_dir + if not _is_path_within(base_dir, link_target): + return False, f"Hardlink escapes extraction directory: {member_name} -> {linkname}" + + return True, "" + + +def _safe_extract_tar(tar: tarfile.TarFile, dest_dir: Path, max_files: int) -> None: + """ + Safely extract a tar archive with full validation. + + Validates all members before extraction and raises HTTPException + if any member fails validation. + + Args: + tar: Open tarfile object + dest_dir: Destination directory for extraction + max_files: Maximum number of files allowed + + Raises: + HTTPException: If archive validation fails + """ + members = tar.getmembers() + + # Check file count + if len(members) > max_files: + raise HTTPException( + status_code=400, + detail={ + "error": f"Archive exceeds maximum file count of {max_files}", + "code": "TOO_MANY_FILES" + } + ) + + # Validate all members before extraction + safe_members = [] + for member in members: + is_valid, error_msg = _validate_tar_member(member, dest_dir) + if not is_valid: + raise HTTPException( + status_code=400, + detail={ + "error": f"Invalid archive: {error_msg}", + "code": "INVALID_ARCHIVE" + } + ) + safe_members.append(member) + + # Extract validated members + tar.extractall(dest_dir, members=safe_members) + + async def deploy_local_agent_logic( body: DeployLocalRequest, current_user: User, @@ -101,28 +245,11 @@ async def deploy_local_agent_logic( temp_dir = Path(tempfile.mkdtemp(prefix="trinity-deploy-")) try: with tarfile.open(fileobj=BytesIO(archive_bytes), mode='r:gz') as tar: - # Security: Check for path traversal - for member in tar.getmembers(): - if member.name.startswith('/') or '..' in member.name: - raise HTTPException( - status_code=400, - detail={ - "error": "Invalid archive: contains path traversal", - "code": "INVALID_ARCHIVE" - } - ) - - # Check file count - if len(tar.getmembers()) > MAX_FILES: - raise HTTPException( - status_code=400, - detail={ - "error": f"Archive exceeds maximum file count of {MAX_FILES}", - "code": "TOO_MANY_FILES" - } - ) - - tar.extractall(temp_dir) + # Security: Safe extraction with full validation + # - Validates paths stay within temp_dir + # - Blocks symlinks/hardlinks pointing outside + # - Rejects device files and FIFOs + _safe_extract_tar(tar, temp_dir, MAX_FILES) except tarfile.TarError as e: raise HTTPException( status_code=400, diff --git a/tests/test_archive_security.py b/tests/test_archive_security.py new file mode 100644 index 000000000..9fc8f89e8 --- /dev/null +++ b/tests/test_archive_security.py @@ -0,0 +1,319 @@ +""" +Archive Security Validation Tests (test_archive_security.py) + +Unit tests for safe tar archive extraction in local agent deployment. +Tests the _validate_tar_member and _safe_extract_tar helpers. + +These tests do NOT require a running backend - they test the validation +logic directly using in-memory tar archives. +""" + +import pytest +import tarfile +import tempfile +import io +import os +from pathlib import Path + +# Import the validation helpers directly +import sys +sys.path.insert(0, str(Path(__file__).parent.parent / "src" / "backend")) + +from services.agent_service.deploy import ( + _is_path_within, + _validate_tar_member, + _safe_extract_tar, +) + + +class TestIsPathWithin: + """Unit tests for _is_path_within helper.""" + + def test_path_within_base(self, tmp_path): + """Path inside base directory returns True.""" + base = tmp_path + target = tmp_path / "subdir" / "file.txt" + assert _is_path_within(base, target) is True + + def test_path_is_base(self, tmp_path): + """Path equal to base directory returns True.""" + base = tmp_path + assert _is_path_within(base, base) is True + + def test_path_outside_base(self, tmp_path): + """Path outside base directory returns False.""" + base = tmp_path / "inside" + base.mkdir() + target = tmp_path / "outside" / "file.txt" + assert _is_path_within(base, target) is False + + def test_path_traversal_rejected(self, tmp_path): + """Path with .. traversal outside base returns False.""" + base = tmp_path / "inside" + base.mkdir() + # This would resolve to tmp_path/file.txt, outside base + target = base / ".." / "file.txt" + assert _is_path_within(base, target) is False + + +class TestValidateTarMember: + """Unit tests for _validate_tar_member helper.""" + + def test_regular_file_allowed(self, tmp_path): + """Regular file with safe path is allowed.""" + member = tarfile.TarInfo(name="file.txt") + member.type = tarfile.REGTYPE + member.size = 100 + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is True + assert error == "" + + def test_nested_file_allowed(self, tmp_path): + """Nested file path is allowed.""" + member = tarfile.TarInfo(name="subdir/nested/file.txt") + member.type = tarfile.REGTYPE + member.size = 100 + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is True + + def test_absolute_path_rejected(self, tmp_path): + """Absolute path is rejected.""" + member = tarfile.TarInfo(name="/etc/passwd") + member.type = tarfile.REGTYPE + member.size = 100 + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "Absolute path" in error + + def test_path_traversal_rejected(self, tmp_path): + """Path with .. traversal is rejected.""" + member = tarfile.TarInfo(name="../outside/file.txt") + member.type = tarfile.REGTYPE + member.size = 100 + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "Path traversal" in error + + def test_nested_traversal_rejected(self, tmp_path): + """Nested path traversal (dir/../../../outside) is rejected.""" + member = tarfile.TarInfo(name="subdir/../../outside.txt") + member.type = tarfile.REGTYPE + member.size = 100 + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "Path traversal" in error + + def test_device_file_rejected(self, tmp_path): + """Character device files are rejected.""" + member = tarfile.TarInfo(name="device") + member.type = tarfile.CHRTYPE + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "Device" in error + + def test_block_device_rejected(self, tmp_path): + """Block device files are rejected.""" + member = tarfile.TarInfo(name="block") + member.type = tarfile.BLKTYPE + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "Device" in error + + def test_fifo_rejected(self, tmp_path): + """FIFO files are rejected.""" + member = tarfile.TarInfo(name="fifo") + member.type = tarfile.FIFOTYPE + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "FIFO" in error + + +class TestValidateTarMemberSymlinks: + """Unit tests for symlink validation in _validate_tar_member.""" + + def test_internal_symlink_allowed(self, tmp_path): + """Symlink pointing to file inside archive is allowed.""" + member = tarfile.TarInfo(name="link.txt") + member.type = tarfile.SYMTYPE + member.linkname = "target.txt" # Points to file in same directory + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is True + + def test_nested_internal_symlink_allowed(self, tmp_path): + """Symlink in subdir pointing to another subdir file is allowed.""" + member = tarfile.TarInfo(name="subdir/link.txt") + member.type = tarfile.SYMTYPE + member.linkname = "../other/target.txt" # Relative, but stays inside + + # Create the structure so resolve works + (tmp_path / "subdir").mkdir(parents=True, exist_ok=True) + (tmp_path / "other").mkdir(parents=True, exist_ok=True) + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is True + + def test_absolute_symlink_rejected(self, tmp_path): + """Symlink with absolute target is rejected.""" + member = tarfile.TarInfo(name="evil_link") + member.type = tarfile.SYMTYPE + member.linkname = "/etc/passwd" + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "Absolute symlink" in error + + def test_escaping_symlink_rejected(self, tmp_path): + """Symlink pointing outside extraction directory is rejected.""" + member = tarfile.TarInfo(name="evil_link") + member.type = tarfile.SYMTYPE + member.linkname = "../../../etc/passwd" + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "escapes" in error.lower() + + +class TestValidateTarMemberHardlinks: + """Unit tests for hardlink validation in _validate_tar_member.""" + + def test_internal_hardlink_allowed(self, tmp_path): + """Hardlink pointing to file inside archive is allowed.""" + member = tarfile.TarInfo(name="hardlink.txt") + member.type = tarfile.LNKTYPE + member.linkname = "original.txt" # Points to file in archive root + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is True + + def test_absolute_hardlink_rejected(self, tmp_path): + """Hardlink with absolute target is rejected.""" + member = tarfile.TarInfo(name="evil_hardlink") + member.type = tarfile.LNKTYPE + member.linkname = "/etc/passwd" + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "Absolute hardlink" in error + + def test_escaping_hardlink_rejected(self, tmp_path): + """Hardlink pointing outside extraction directory is rejected.""" + member = tarfile.TarInfo(name="evil_hardlink") + member.type = tarfile.LNKTYPE + member.linkname = "../../../etc/passwd" + + is_valid, error = _validate_tar_member(member, tmp_path) + assert is_valid is False + assert "escapes" in error.lower() + + +class TestSafeExtractTar: + """Integration tests for _safe_extract_tar with real archives.""" + + def _create_tar_with_file(self, name: str, content: bytes) -> io.BytesIO: + """Create a tar archive with a single file.""" + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode='w:gz') as tar: + info = tarfile.TarInfo(name=name) + info.size = len(content) + tar.addfile(info, io.BytesIO(content)) + buffer.seek(0) + return buffer + + def _create_tar_with_symlink(self, link_name: str, target: str) -> io.BytesIO: + """Create a tar archive with a symlink.""" + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode='w:gz') as tar: + # Add a target file first + content = b"target content" + target_info = tarfile.TarInfo(name="target.txt") + target_info.size = len(content) + tar.addfile(target_info, io.BytesIO(content)) + + # Add the symlink + link_info = tarfile.TarInfo(name=link_name) + link_info.type = tarfile.SYMTYPE + link_info.linkname = target + tar.addfile(link_info) + buffer.seek(0) + return buffer + + def test_extract_safe_archive(self, tmp_path): + """Safe archive extracts successfully.""" + buffer = self._create_tar_with_file("test.txt", b"hello world") + + with tarfile.open(fileobj=buffer, mode='r:gz') as tar: + _safe_extract_tar(tar, tmp_path, max_files=100) + + assert (tmp_path / "test.txt").exists() + assert (tmp_path / "test.txt").read_bytes() == b"hello world" + + def test_extract_internal_symlink_allowed(self, tmp_path): + """Archive with internal symlink extracts successfully.""" + buffer = self._create_tar_with_symlink("link.txt", "target.txt") + + with tarfile.open(fileobj=buffer, mode='r:gz') as tar: + _safe_extract_tar(tar, tmp_path, max_files=100) + + assert (tmp_path / "target.txt").exists() + assert (tmp_path / "link.txt").is_symlink() + + def test_extract_escaping_symlink_rejected(self, tmp_path): + """Archive with escaping symlink raises HTTPException.""" + buffer = self._create_tar_with_symlink("evil.txt", "../../../etc/passwd") + + from fastapi import HTTPException + + with tarfile.open(fileobj=buffer, mode='r:gz') as tar: + with pytest.raises(HTTPException) as exc_info: + _safe_extract_tar(tar, tmp_path, max_files=100) + + assert exc_info.value.status_code == 400 + assert "INVALID_ARCHIVE" in str(exc_info.value.detail) + + def test_extract_too_many_files_rejected(self, tmp_path): + """Archive with too many files raises HTTPException.""" + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode='w:gz') as tar: + for i in range(10): + content = f"file {i}".encode() + info = tarfile.TarInfo(name=f"file_{i}.txt") + info.size = len(content) + tar.addfile(info, io.BytesIO(content)) + buffer.seek(0) + + from fastapi import HTTPException + + with tarfile.open(fileobj=buffer, mode='r:gz') as tar: + with pytest.raises(HTTPException) as exc_info: + _safe_extract_tar(tar, tmp_path, max_files=5) # Only allow 5 + + assert exc_info.value.status_code == 400 + assert "TOO_MANY_FILES" in str(exc_info.value.detail) + + def test_extract_path_traversal_rejected(self, tmp_path): + """Archive with path traversal raises HTTPException.""" + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode='w:gz') as tar: + content = b"malicious" + info = tarfile.TarInfo(name="../escape.txt") + info.size = len(content) + tar.addfile(info, io.BytesIO(content)) + buffer.seek(0) + + from fastapi import HTTPException + + with tarfile.open(fileobj=buffer, mode='r:gz') as tar: + with pytest.raises(HTTPException) as exc_info: + _safe_extract_tar(tar, tmp_path, max_files=100) + + assert exc_info.value.status_code == 400 + assert "INVALID_ARCHIVE" in str(exc_info.value.detail)