diff --git a/netbox/core/data_backends.py b/netbox/core/data_backends.py index 11f9900b5..c85d81715 100644 --- a/netbox/core/data_backends.py +++ b/netbox/core/data_backends.py @@ -1,9 +1,8 @@ import logging -import os import re import tempfile from contextlib import contextmanager -from pathlib import Path +from pathlib import Path, PurePosixPath from urllib.parse import urlparse from django import forms @@ -171,6 +170,7 @@ class S3Backend(DataBackend): import boto3 local_path = tempfile.TemporaryDirectory() + local_root = Path(local_path.name).resolve() # Initialize the S3 resource and bucket aws_access_key_id = self.params.get('aws_access_key_id') @@ -185,16 +185,31 @@ class S3Backend(DataBackend): ) bucket = s3.Bucket(self._bucket_name) - # Download all files within the specified path - for obj in bucket.objects.filter(Prefix=self._remote_path): - local_filename = os.path.join(local_path.name, obj.key) - # Build local path - Path(os.path.dirname(local_filename)).mkdir(parents=True, exist_ok=True) - bucket.download_file(obj.key, local_filename) + try: + # Download all files within the specified path + for obj in bucket.objects.filter(Prefix=self._remote_path): + local_filename = self._resolve_local_path(local_root, obj.key) + # Build local path + local_filename.parent.mkdir(parents=True, exist_ok=True) + bucket.download_file(obj.key, str(local_filename)) - yield local_path.name + yield local_path.name + finally: + local_path.cleanup() - local_path.cleanup() + @staticmethod + def _resolve_local_path(local_root, key): + # S3 object keys are POSIX-style paths. Strip any leading separator so the key + # joins onto the temp directory rather than replacing it, then ensure the + # resolved destination remains within the temp directory to prevent path + # traversal via crafted object keys. + key_parts = PurePosixPath(key.lstrip('/')).parts + local_filename = local_root.joinpath(*key_parts).resolve() + if not local_filename.is_relative_to(local_root) or local_filename == local_root: + raise SyncError( + _("Invalid S3 object key '{key}': resolves outside of the local data directory").format(key=key) + ) + return local_filename @property def _region_name(self): diff --git a/netbox/core/tests/test_data_backends.py b/netbox/core/tests/test_data_backends.py index 51cae6231..9f96a0cb3 100644 --- a/netbox/core/tests/test_data_backends.py +++ b/netbox/core/tests/test_data_backends.py @@ -1,9 +1,12 @@ +import tempfile +from pathlib import Path from unittest import skipIf from unittest.mock import patch from django.test import TestCase -from core.data_backends import url_has_embedded_credentials +from core.data_backends import S3Backend, url_has_embedded_credentials +from core.exceptions import SyncError try: import dulwich # noqa: F401 @@ -114,3 +117,49 @@ class GitBackendCredentialIntegrationTestCase(TestCase): self.assertEqual(kwargs.get('username'), None) self.assertEqual(kwargs.get('password'), None) + + +class S3BackendKeyResolutionTestCase(TestCase): + """ + Verify that S3Backend resolves object keys to local paths safely, rejecting any + key that would write outside of the local temporary directory. + """ + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.local_root = Path(self._tmp.name).resolve() + + def test_simple_key_resolves_under_root(self): + result = S3Backend._resolve_local_path(self.local_root, 'data/file.yaml') + self.assertEqual(result, self.local_root / 'data' / 'file.yaml') + + def test_nested_key_resolves_under_root(self): + result = S3Backend._resolve_local_path(self.local_root, 'a/b/c/d.txt') + self.assertEqual(result, self.local_root / 'a' / 'b' / 'c' / 'd.txt') + + def test_absolute_key_is_contained(self): + # A key beginning with '/' must not escape the local root + result = S3Backend._resolve_local_path(self.local_root, '/etc/passwd') + self.assertEqual(result, self.local_root / 'etc' / 'passwd') + self.assertTrue(result.is_relative_to(self.local_root)) + + def test_parent_traversal_is_rejected(self): + with self.assertRaises(SyncError): + S3Backend._resolve_local_path(self.local_root, '../escape.txt') + + def test_nested_parent_traversal_is_rejected(self): + with self.assertRaises(SyncError): + S3Backend._resolve_local_path(self.local_root, 'foo/../../escape.txt') + + def test_absolute_parent_traversal_is_rejected(self): + with self.assertRaises(SyncError): + S3Backend._resolve_local_path(self.local_root, '/../escape.txt') + + def test_empty_key_is_rejected(self): + with self.assertRaises(SyncError): + S3Backend._resolve_local_path(self.local_root, '') + + def test_root_key_is_rejected(self): + with self.assertRaises(SyncError): + S3Backend._resolve_local_path(self.local_root, '/')