Use safe_load to avoid code invocation from YAML files

connexion/api.py

@@ -36,6 +36,9 @@
 
 def compatibility_layer(spec):
     """Make specs compatible with older versions of Connexion."""
+    if not isinstance(spec, dict):
+        return spec
+
     # Make all response codes be string
     for path_name, methods_available in spec.get('paths', {}).items():
         for method_name, method_def in methods_available.items():
@@ -92,7 +95,7 @@ def __init__(self, swagger_yaml_path, base_url=None, arguments=None,
                 swagger_template = contents.decode('utf-8', 'replace')
 
             swagger_string = jinja2.Template(swagger_template).render(**arguments)
-            self.specification = yaml.load(swagger_string)  # type: dict
+            self.specification = yaml.safe_load(swagger_string)  # type: dict
 
         logger.debug('Read specification', extra={'spec': self.specification})
 

tests/test_api.py

@@ -3,10 +3,10 @@
 import pathlib
 import tempfile
 
+import pytest
 from connexion.api import Api
 from swagger_spec_validator.common import SwaggerValidationError
-
-import pytest
+from yaml import YAMLError
 
 TEST_FOLDER = pathlib.Path(__file__).parent
 
@@ -53,3 +53,14 @@ def test_invalid_encoding():
         f.write(u"swagger: '2.0'\ninfo:\n  title: Foo 整\n  version: v1\npaths: {}".encode('gbk'))
         f.flush()
         Api(pathlib.Path(f.name), "/api/v1.0")
+
+
+def test_use_of_safe_load_for_yaml_swagger_specs():
+    with pytest.raises(YAMLError):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write('!!python/object:object {}\n'.encode())
+            f.flush()
+            try:
+                Api(pathlib.Path(f.name), "/api/v1.0")
+            except SwaggerValidationError:
+                pytest.fail("Could load invalid YAML file, use yaml.safe_load!")