CVE-2023-41319 (Python, published 2023-09-06T18:15Z)
Description: value: Fides is an open-source privacy engineering platform for managing the fulfillment of data privacy requests in a runtime environment, and the enforcement of privacy regulations in code. The Fides webserver API allows custom integrations to be uploaded as a ZIP file. This ZIP file must contain YAML files, but Fides can be configured to also accept the inclusion of custom Python code in it. The custom code is executed in a restricted, sandboxed environment, but the sandbox can be bypassed to execute any arbitrary code. The vulnerability allows the execution of arbitrary code on the target system within the context of the webserver python process owner on the webserver container, which by default is `root`, and leverage that access to attack underlying infrastructure and integrated systems. This vulnerability affects Fides versions `2.11.0` through `2.19.0`. Exploitation is limited to API clients with the `CONNECTOR_TEMPLATE_REGISTER` authorization scope. In the Fides Admin UI this scope is restricted to highly privileged users, specifically root users and users with the owner role.  Exploitation is only possible if the security configuration parameter `allow_custom_connector_functions` is enabled by the user deploying the Fides webserver container, either in `fides.toml` or by setting the env var `FIDES__SECURITY__ALLOW_CUSTOM_CONNECTOR_FUNCTIONS=True`. By default this configuration parameter is disabled. The vulnerability has been patched in Fides version `2.19.0`. Users are advised to upgrade to this version or later to secure their systems against this threat. Users unable to upgrade should ensure that `allow_custom_connector_functions` in `fides.toml` and the `FIDES__SECURITY__ALLOW_CUSTOM_CONNECTOR_FUNCTIONS` are both either unset or explicit set to `False`.
Severity: nan (CVSS v3 base score: 7.2)
Weakness: CWE-94 \u2014 Improper Control of Generation of Code ('Code Injection')
The product constructs all or part of a code segment using externally-influenced input from an upstream component, but it does not neutralize or incorrectly neutralizes special elements that could modify the syntax or behavior of the intended code segment.
Repository: https://github.com/ethyca/fides
Files changed:
clients/admin-ui/src/features/connector-templates/ConnectorTemplateUploadModal.tsx
docker-compose.yml
requirements.txt
src/fides/api/schemas/saas/connector_template.py
src/fides/api/service/connectors/saas/connector_registry_service.py
src/fides/config/security_settings.py
tests/ops/api/v1/endpoints/test_saas_config_endpoints.py
tests/ops/models/test_custom_connector_template.py
tests/ops/service/connectors/test_connector_template_loaders.py
Commit message: Merge pull request from GHSA-p6p2-qq95-vq5h

* Removing custom connector template functions

* Updating connector template modal description

* Adding function upload tests

Vulnerable code:
            dataset, but may also contain an icon (.svg) and custom functions
            (.py) as optional files.
      FIDES__SECURITY__ALLOW_CUSTOM_CONNECTOR_FUNCTIONS: "True"
AccessControl==6.0
RestrictedPython==6.0.0
    SaaS connector (SaaS config, dataset, icon, functions, etc.)
    functions: Optional[str]
from ast import AST, AnnAssign
from operator import getitem
from typing import Any, Dict, Iterable, List, Optional, Tuple, Type
from AccessControl.ZopeGuards import safe_builtins
from RestrictedPython import compile_restricted
from RestrictedPython.transformer import RestrictingNodeTransformer
from fides.api.common_exceptions import FidesopsException, ValidationError
from fides.config import CONFIG
                        functions=None,
        Registers a custom connector template by converting it to a ConnectorTemplate,
        registering any custom functions, and adding it to the loader's template dictionary.
            functions=template.functions,
        # register custom functions if available
        if template.functions:
            register_custom_functions(template.functions)
            logger.info(
                f"Loaded functions from the custom connector template '{template.key}'"
            )

            elif info.filename.endswith(".py"):
                if not function_contents:
                    function_contents = file_contents
                else:
                    raise ValidationError(
                        "Multiple Python (.py) files found, only one is allowed."
                    )
            functions=function_contents,


def register_custom_functions(script: str) -> None:
    """
    Registers custom functions by executing the given script in a restricted environment.

    The script is compiled and executed with RestrictedPython, which is designed to reduce
    the risk of executing untrusted code. It provides a set of safe builtins to prevent
    malicious or unintended behavior.

    Args:
        script (str): The Python script containing the custom functions to be registered.

    Raises:
        FidesopsException: If allow_custom_connector_functions is disabled.
        SyntaxError: If the script contains a syntax error or uses restricted language features.
        Exception: If an exception occurs during the execution of the script.
    """

    if CONFIG.security.allow_custom_connector_functions:
        restricted_code = compile_restricted(
            script, "<string>", "exec", policy=CustomRestrictingNodeTransformer
        )
        safe_builtins["__import__"] = custom_guarded_import
        safe_builtins["_getitem_"] = getitem
        safe_builtins["staticmethod"] = staticmethod

        # pylint: disable=exec-used
        exec(
            restricted_code,
            {
                "__metaclass__": type,
                "__name__": "restricted_module",
                "__builtins__": safe_builtins,
            },
        )
    else:
        raise FidesopsException(
            message="The import of connector templates with custom functions is disabled by the 'security.allow_custom_connector_functions' setting."
        )


class CustomRestrictingNodeTransformer(RestrictingNodeTransformer):
    """
    Custom node transformer class that extends RestrictedPython's RestrictingNodeTransformer
    to allow the use of type annotations (AnnAssign) in restricted code.
    """

    def visit_AnnAssign(self, node: AnnAssign) -> AST:
        return self.node_contents_visit(node)


def custom_guarded_import(
    name: str,
    _globals: Optional[dict] = None,
    _locals: Optional[dict] = None,
    fromlist: Optional[Tuple[str, ...]] = None,
    level: int = 0,
) -> Any:
    """
    A custom import function that prevents the import of certain potentially unsafe modules.
    """
    if name in [
        "os",
        "sys",
        "subprocess",
        "shutil",
        "socket",
        "importlib",
        "tempfile",
        "glob",
    ]:
        # raising SyntaxError to be consistent with exceptions thrown from other guarded functions
        raise SyntaxError(f"Import of '{name}' module is not allowed.")
    if fromlist is None:
        fromlist = ()
    return __import__(name, _globals, _locals, fromlist, level)
    allow_custom_connector_functions: Optional[bool] = Field(
        default=False,
        description="Enables or disables the ability to import connector templates with custom functions. When enabled, custom functions which will be loaded in a restricted environment to minimize security risks.",
    )
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
                "icon.svg": planet_express_icon,
            }
        )

    @pytest.fixture
    def connector_template_no_functions(
        self,
        planet_express_config,
        planet_express_dataset,
        planet_express_icon,
    ):
        return create_zip_file(
            {
                "config.yml": planet_express_config,
                "dataset.yml": planet_express_dataset,
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
                "icon.svg": planet_express_icon,
            }
        )

    @pytest.fixture
    def connector_template_duplicate_functions(
        self,
        planet_express_config,
        planet_express_dataset,
        planet_express_functions,
        planet_express_icon,
    ):
        return create_zip_file(
            {
                "config.yml": planet_express_config,
                "dataset.yml": planet_express_dataset,
                "1_functions.py": planet_express_functions,
                "2_functions.py": planet_express_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        CONFIG.security.allow_custom_connector_functions = True
            (
                "connector_template_no_functions",
                200,
                {"message": "Connector template successfully registered."},
            ),
            (
                "connector_template_duplicate_functions",
                400,
                {"detail": "Multiple Python (.py) files found, only one is allowed."},
            ),
    @mock.patch(
        "fides.api.service.connectors.saas.connector_registry_service.register_custom_functions"
    )  # prevent functions from being registered to avoid test conflicts
    def test_register_connector_template_allow_custom_connector_functions(
        self,
        mock_register_custom_functions: MagicMock,
        api_client: TestClient,
        register_connector_template_url,
        generate_auth_header,
        zip_file,
        status_code,
        details,
        request,
    ):
        CONFIG.security.allow_custom_connector_functions = True
        auth_header = generate_auth_header(scopes=[CONNECTOR_TEMPLATE_REGISTER])
        response = api_client.post(
            register_connector_template_url,
            headers=auth_header,
            files={
                "file": (
                    "template.zip",
                    request.getfixturevalue(zip_file).read(),
                    "application/zip",
                )
            },
        )
        assert response.status_code == status_code
        assert response.json() == details

    @pytest.mark.parametrize(
        "zip_file, status_code, details",
        [
            (
                "complete_connector_template",
                400,
                {
                    "detail": "The import of connector templates with custom functions is disabled by the 'security.allow_custom_connector_functions' setting."
                },
            ),
            (
                "connector_template_no_functions",
                200,
                {"message": "Connector template successfully registered."},
            ),
        ],
    )
    def test_register_connector_template_disallow_custom_connector_functions(
        CONFIG.security.allow_custom_connector_functions = False
        planet_express_functions,
            functions=planet_express_functions,
        assert custom_connector.functions == planet_express_functions
    register_custom_functions,
        planet_express_functions,
                "functions.py": planet_express_functions,
        CONFIG.security.allow_custom_connector_functions = True

        planet_express_functions,
        CONFIG.security.allow_custom_connector_functions = True

                functions=planet_express_functions,
            )
        ]

        # verify the custom functions aren't loaded if the template is invalid
        connector_templates = CustomConnectorTemplateLoader.get_connector_templates()
        assert connector_templates == {}

        with pytest.raises(NoSuchSaaSRequestOverrideException):
            SaaSRequestOverrideFactory.get_override(
                "planet_express_user_access", SaaSRequestType.READ
            )

        # assert the strategy was not registered
        authentication_strategies = AuthenticationStrategy.get_strategies()
        assert "planet_express" not in [
            strategy.name for strategy in authentication_strategies
        ]

    @mock.patch(
        "fides.api.models.custom_connector_template.CustomConnectorTemplate.all"
    )
    def test_custom_connector_template_loader_invalid_functions(
        self,
        mock_all: MagicMock,
        planet_express_config,
        planet_express_dataset,
        planet_express_icon,
    ):
        CONFIG.security.allow_custom_connector_functions = True

        # save custom connector template to the database
        mock_all.return_value = [
            CustomConnectorTemplate(
                key="planet_express",
                name="Planet Express",
                config=planet_express_config,
                dataset=planet_express_dataset,
                icon=planet_express_icon,
                functions="planet_express_functions",
            )
        ]

        # verify nothing is loaded if the custom functions fail to load
        connector_templates = CustomConnectorTemplateLoader.get_connector_templates()
        assert connector_templates == {}

    @mock.patch(
        "fides.api.models.custom_connector_template.CustomConnectorTemplate.all"
    )
    def test_custom_connector_template_loader_custom_connector_functions_disabled(
        self,
        mock_all: MagicMock,
        planet_express_config,
        planet_express_dataset,
        planet_express_icon,
        planet_express_functions,
    ):
        CONFIG.security.allow_custom_connector_functions = False

        mock_all.return_value = [
            CustomConnectorTemplate(
                key="planet_express",
                name="Planet Express",
                config=planet_express_config,
                dataset=planet_express_dataset,
                icon=planet_express_icon,
                functions=planet_express_functions,
        # load custom connector templates from the database
        with pytest.raises(NoSuchSaaSRequestOverrideException):
            SaaSRequestOverrideFactory.get_override(
                "planet_express_user_access", SaaSRequestType.READ
            )

        # assert the strategy was not registered
        authentication_strategies = AuthenticationStrategy.get_strategies()
        assert "planet_express" not in [
            strategy.name for strategy in authentication_strategies
        ]

    @mock.patch(
        "fides.api.models.custom_connector_template.CustomConnectorTemplate.all"
    )
    def test_custom_connector_template_loader_custom_connector_functions_disabled_custom_functions(
        self,
        mock_all: MagicMock,
        planet_express_config,
        planet_express_dataset,
        planet_express_icon,
    ):
        """
        A connector template with no custom functions should still be loaded
        even if allow_custom_connector_functions is set to false
        """

        CONFIG.security.allow_custom_connector_functions = False

        # save custom connector template to the database
        mock_all.return_value = [
            CustomConnectorTemplate(
                key="planet_express",
                name="Planet Express",
                config=planet_express_config,
                dataset=planet_express_dataset,
                icon=planet_express_icon,
                functions=None,
            )
        ]

        # load custom connector templates from the database
        connector_templates = CustomConnectorTemplateLoader.get_connector_templates()
        assert connector_templates == {
            "planet_express": ConnectorTemplate(
                config=planet_express_config,
                dataset=planet_express_dataset,
                icon=planet_express_icon,
                human_readable="Planet Express",
            )
        }

        planet_express_functions,
        CONFIG.security.allow_custom_connector_functions = True

                functions=planet_express_functions,
                functions=planet_express_functions,
        # assert the request override was registered
        SaaSRequestOverrideFactory.get_override(
            "planet_express_user_access", SaaSRequestType.READ
        )

        # assert the strategy was registered
        authentication_strategies = AuthenticationStrategy.get_strategies()
        assert "planet_express" in [
            strategy.name for strategy in authentication_strategies
        ]

        planet_express_functions,
        CONFIG.security.allow_custom_connector_functions = True

                functions=planet_express_functions,
        planet_express_functions,
                        "functions.py": planet_express_functions,
                        "functions.py": planet_express_functions,
    def test_custom_connector_template_loader_disallowed_modules(
        CONFIG.security.allow_custom_connector_functions = True

        with pytest.raises(SyntaxError) as exc:
            CustomConnectorTemplateLoader.save_template(
                MagicMock(),
                ZipFile(
                    create_zip_file(
                        {
                            "config.yml": planet_express_config,
                            "dataset.yml": planet_express_dataset,
                            "functions.py": "import os",
                            "icon.svg": planet_express_icon,
                        }
                    )
                ),
        assert "Import of 'os' module is not allowed." == str(exc.value)


class TestRegisterCustomFunctions:
    def test_function_loader(self):
        """Verify that all override implementations can be loaded by RestrictedPython"""

        overrides_path = "src/fides/api/service/saas_request/override_implementations"

        for filename in os.listdir(overrides_path):
            if filename.endswith(".py") and filename != "__init__.py":
                file_path = os.path.join(overrides_path, filename)
                with open(file_path, "r") as file:
                    register_custom_functions(file.read())

Fixed code:
            dataset, but may also contain an icon (.svg) as an optional file.
    SaaS connector (SaaS config, dataset, icon, etc.)
from typing import Dict, Iterable, List, Optional, Type
from fides.api.common_exceptions import ValidationError
        Registers a custom connector template by converting it to a ConnectorTemplate
        and adding it to the loader's template dictionary.
    def test_register_connector_template(
    @mock.patch(
        "fides.api.models.custom_connector_template.CustomConnectorTemplate.create_or_update"
    )
    def test_custom_connector_save_template_with_functions(
        mock_create_or_update: MagicMock,
        planet_express_functions,
        db = MagicMock()

        CustomConnectorTemplateLoader.save_template(
            db,
            ZipFile(
                create_zip_file(
                    {
                        "config.yml": planet_express_config,
                        "dataset.yml": planet_express_dataset,
                        "functions.py": planet_express_functions,
                        "icon.svg": planet_express_icon,
                    }
                )
            ),
        )

        # assert the request override was ignored
        with pytest.raises(NoSuchSaaSRequestOverrideException) as exc:
            SaaSRequestOverrideFactory.get_override(
                "planet_express_user_access", SaaSRequestType.UPDATE
        assert (
            f"Custom SaaS override 'planet_express_user_access' does not exist."
            in str(exc.value)
        )

        # assert the strategy was ignored
        authentication_strategies = AuthenticationStrategy.get_strategies()
        assert "planet_express" not in [
            strategy.name for strategy in authentication_strategies
        ]