Restrict pickle class loading for .localstate deserialization

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)
 --------------------

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'

ebcli/operations/localops.py

@@ -10,15 +10,30 @@
 # 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
 from ebcli.core import fileoperations, io
 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

tests/unit/operations/test_localops.py

@@ -66,18 +66,42 @@ 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):
         loads.return_value = self.localstate
 
         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)