diff --git a/CHANGES.rst b/CHANGES.rst index 33e9ee2e2..2b50a0fbb 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -1,5 +1,10 @@ Changelog ========= +-------------------- +3.27.2 (2026-05-13) +-------------------- +- Restrict pickle class loading for ``.localstate`` deserialization in ``eb local`` commands + -------------------- 3.27.1 (2026-03-25) -------------------- diff --git a/ebcli/__init__.py b/ebcli/__init__.py index c49d75942..3e0792f74 100644 --- a/ebcli/__init__.py +++ b/ebcli/__init__.py @@ -11,4 +11,4 @@ # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. -__version__ = '3.27.1' +__version__ = '3.27.2' diff --git a/ebcli/operations/localops.py b/ebcli/operations/localops.py index f7c8c27e8..87da6a22c 100644 --- a/ebcli/operations/localops.py +++ b/ebcli/operations/localops.py @@ -10,7 +10,8 @@ # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. -from botocore.compat import six +import io as _io +import pickle from ebcli.containers.envvarcollector import EnvvarCollector from ebcli.containers.pathconfig import PathConfig @@ -18,7 +19,21 @@ from ebcli.lib import utils from ebcli.operations import commonops, envvarops from ebcli.resources.strings import strings -cPickle = six.moves.cPickle + + +class _SafeUnpickler(pickle.Unpickler): + """Only allow deserializing LocalState and EnvvarCollector.""" + _ALLOWED = { + ('ebcli.operations.localops', 'LocalState'), + ('ebcli.containers.envvarcollector', 'EnvvarCollector'), + ('__builtin__', 'set'), + ('builtins', 'set'), + } + + def find_class(self, module, name): + if (module, name) in self._ALLOWED: + return super().find_class(module, name) + raise pickle.UnpicklingError(f"Forbidden: {module}.{name}") class LocalState(object): @@ -30,15 +45,15 @@ def __init__(self, envvarcollector): self.envvarcollector = envvarcollector def dumps(self, path): - fileoperations.write_to_data_file(data=cPickle.dumps(self, protocol=2), + fileoperations.write_to_data_file(data=pickle.dumps(self, protocol=2), location=path) @classmethod def loads(cls, path): try: data = fileoperations.read_from_data_file(path) - return cPickle.loads(data) - except IOError: + return _SafeUnpickler(_io.BytesIO(data)).load() + except (IOError, pickle.UnpicklingError): return cls(EnvvarCollector()) @classmethod diff --git a/tests/unit/operations/test_localops.py b/tests/unit/operations/test_localops.py index f797c8808..e089fb4be 100644 --- a/tests/unit/operations/test_localops.py +++ b/tests/unit/operations/test_localops.py @@ -66,14 +66,13 @@ def setUp(self): def test_constructor(self): self.assertEqual(self.envvarcollector, self.localstate.envvarcollector) - @patch('ebcli.operations.localops.cPickle') + @patch('ebcli.operations.localops._SafeUnpickler') @patch('ebcli.operations.localops.fileoperations') - def test_loads(self, fileoperations, cPickle): + def test_loads(self, fileoperations, safe_unpickler): fileoperations.read_from_data_file.return_value = b'foo' - cPickle.loads.return_value = self.localstate + safe_unpickler.return_value.load.return_value = self.localstate self.assertEqual(self.localstate, LocalState.loads(LOCAL_STATE_PATH)) - cPickle.loads.assert_called_once_with(b'foo') @patch('ebcli.operations.localops.LocalState.loads') def test_get_envvarcollector(self, loads): @@ -81,3 +80,28 @@ def test_get_envvarcollector(self, loads): self.assertEqual(self.envvarcollector, LocalState.get_envvarcollector(LOCAL_STATE_PATH)) + + @patch('ebcli.operations.localops.fileoperations') + def test_loads_rejects_unlisted_class(self, fileoperations): + """Verify that a pickle containing a class not in the allowlist is rejected.""" + import pickle + fileoperations.read_from_data_file.return_value = pickle.dumps( + object(), protocol=2) + + result = LocalState.loads(LOCAL_STATE_PATH) + + self.assertIsInstance(result, LocalState) + self.assertDictEqual({}, result.envvarcollector.map) + + def test_dumps_loads_roundtrip(self): + """Verify dumps/loads round-trip through real pickle.""" + import os, tempfile + tmpdir = tempfile.mkdtemp() + path = os.path.join(tmpdir, '.localstate') + try: + self.localstate.dumps(path) + result = LocalState.loads(path) + self.assertDictEqual(self.envvarcollector.map, result.envvarcollector.map) + finally: + os.remove(path) + os.rmdir(tmpdir)