allow commands from settings

Author: Richardk2n

README.rst

@@ -78,6 +78,19 @@ Configuration
      - ``normal``, ``silent``, ``skip`` or ``error``
      - ``mypy`` **parameter** ``follow-imports``. In ``mypy`` this is ``normal`` by default. We set it ``silent``, to sort out unwanted results. This can cause cache invalidation if you also run ``mypy`` in other ways. Setting this to ``normal`` avoids this at the cost of a small performance penalty.
      - ``silent``
+   * - ``mypy_command``
+     - ``pylsp.plugins.pylsp_mypy.mypy_command``
+     - ``array`` of ``string`` items
+     - **The command to run mypy**. This is useful if you want to run mypy in a specific virtual environment. Requires env variable ``PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION`` to be set.
+     - ``[]``
+   * - ``dmypy_command``
+     - ``pylsp.plugins.pylsp_mypy.dmypy_command``
+     - ``array`` of ``string`` items
+     - **The command to run dmypy**. This is useful if you want to run dmypy in a specific virtual environment. Requires env variable ``PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION`` to be set.
+     - ``[]``
+
+Both ``mypy_command`` and ``dmypy_command`` could be used by a malicious repo to execute arbitrary code by looking at its source with this plugin active.
+Still users want this feature. For security reasons this is disabled by default. If you really want it and accept the risks set the environment variable ``PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION`` in order to activate it.
 
 Using a ``pyproject.toml`` for configuration, which is in fact the preferred way, your configuration could look like this:
 
@@ -151,6 +164,26 @@ With ``report_progress`` your config could look like this:
         "report_progress": True
     }
 
+With ``mypy_command`` your config could look like this:
+
+::
+
+    {
+        "enabled": True,
+        "mypy_command": ["poetry", "run", "mypy"]
+    }
+
+With ``dmypy_command`` your config could look like this:
+
+::
+
+    {
+        "enabled": True,
+        "live_mode": False,
+        "dmypy": True,
+        "dmypy_command": ["/path/to/venv/bin/dmypy"]
+    }
+
 Developing
 -------------
 

pylsp_mypy/plugin.py

@@ -165,6 +165,30 @@ def match_exclude_patterns(document_path: str, exclude_patterns: list) -> bool:
     return False
 
 
+def get_cmd(settings: Dict[str, Any], cmd: str) -> List[str]:
+    """
+    Get the command to run from settings, falling back to searching the PATH.
+    If the command is not found in the settings and is not available on the PATH, an
+    empty list is returned.
+    """
+    command_key = f"{cmd}_command"
+    command: List[str] = settings.get(command_key, [])
+
+    if not (command and os.getenv("PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION")):
+        # env var is required to allow command from settings
+        if shutil.which(cmd):  # Fallback to PATH
+            log.debug(
+                f"'{command_key}' not found in settings or not allowed, using '{cmd}' from PATH"
+            )
+            command = [cmd]
+        else:  # Fallback to API
+            command = []
+
+    log.debug(f"Using {cmd} command: {command}")
+
+    return command
+
+
 @hookimpl
 def pylsp_lint(
     config: Config, workspace: Workspace, document: Document, is_saved: bool
@@ -304,18 +328,21 @@ def get_diagnostics(
         args.extend(["--incremental", "--follow-imports", settings.get("follow-imports", "silent")])
         args = apply_overrides(args, overrides)
 
-        if shutil.which("mypy"):
-            # mypy exists on path
-            # -> use mypy on path
+        mypy_command: List[str] = get_cmd(settings, "mypy")
+
+        if mypy_command:
+            # mypy exists on PATH or was provided by settings
+            # -> use this mypy
             log.info("executing mypy args = %s on path", args)
             completed_process = subprocess.run(
-                ["mypy", *args], capture_output=True, **windows_flag, encoding="utf-8"
+                [*mypy_command, *args], capture_output=True, **windows_flag, encoding="utf-8"
             )
             report = completed_process.stdout
             errors = completed_process.stderr
             exit_status = completed_process.returncode
         else:
-            # mypy does not exist on path, but must exist in the env pylsp-mypy is installed in
+            # mypy does not exist on PATH and was not provided by settings,
+            # but must exist in the env pylsp-mypy is installed in
             # -> use mypy via api
             log.info("executing mypy args = %s via api", args)
             report, errors, exit_status = mypy_api.run(args)
@@ -326,11 +353,13 @@ def get_diagnostics(
         # If daemon is dead/absent, kill will no-op.
         # In either case, reset to fresh state
 
-        if shutil.which("dmypy"):
-            # dmypy exists on path
-            # -> use dmypy on path
+        dmypy_command: List[str] = get_cmd(settings, "dmypy")
+
+        if dmypy_command:
+            # dmypy exists on PATH or was provided by settings
+            # -> use this dmypy
             completed_process = subprocess.run(
-                ["dmypy", "--status-file", dmypy_status_file, "status"],
+                [*dmypy_command, "--status-file", dmypy_status_file, "status"],
                 capture_output=True,
                 **windows_flag,
                 encoding="utf-8",
@@ -350,7 +379,8 @@ def get_diagnostics(
                     encoding="utf-8",
                 )
         else:
-            # dmypy does not exist on path, but must exist in the env pylsp-mypy is installed in
+            # dmypy does not exist on PATH and was not provided by settings,
+            # but must exist in the env pylsp-mypy is installed in
             # -> use dmypy via api
             _, errors, exit_status = mypy_api.run_dmypy(
                 ["--status-file", dmypy_status_file, "status"]
@@ -365,18 +395,19 @@ def get_diagnostics(
 
         # run to use existing daemon or restart if required
         args = ["--status-file", dmypy_status_file, "run", "--"] + apply_overrides(args, overrides)
-        if shutil.which("dmypy"):
-            # dmypy exists on path
-            # -> use mypy on path
+        if dmypy_command:
+            # dmypy exists on PATH or was provided by settings
+            # -> use this dmypy
             log.info("dmypy run args = %s via path", args)
             completed_process = subprocess.run(
-                ["dmypy", *args], capture_output=True, **windows_flag, encoding="utf-8"
+                [*dmypy_command, *args], capture_output=True, **windows_flag, encoding="utf-8"
             )
             report = completed_process.stdout
             errors = completed_process.stderr
             exit_status = completed_process.returncode
         else:
-            # dmypy does not exist on path, but must exist in the env pylsp-mypy is installed in
+            # dmypy does not exist on PATH and was not provided by settings,
+            # but must exist in the env pylsp-mypy is installed in
             # -> use dmypy via api
             log.info("dmypy run args = %s via api", args)
             report, errors, exit_status = mypy_api.run_dmypy(args)

test/test_plugin.py

@@ -383,3 +383,103 @@ def test_config_exclude(tmpdir, workspace):
     workspace.update_config({"pylsp": {"plugins": {"pylsp_mypy": {"exclude": [exclude_path]}}}})
     diags = plugin.pylsp_lint(workspace._config, workspace, doc, is_saved=False)
     assert diags == []
+
+
+@pytest.mark.parametrize(
+    ("command", "settings", "cmd_on_path", "environmentVariableSet", "expected"),
+    [
+        ("mypy", {}, ["/bin/mypy"], True, ["mypy"]),
+        ("mypy", {}, None, True, []),
+        ("mypy", {"mypy_command": ["/path/to/mypy"]}, "/bin/mypy", True, ["/path/to/mypy"]),
+        ("mypy", {"mypy_command": ["/path/to/mypy"]}, None, True, ["/path/to/mypy"]),
+        ("dmypy", {}, "/bin/dmypy", True, ["dmypy"]),
+        ("dmypy", {}, None, True, []),
+        ("dmypy", {"dmypy_command": ["/path/to/dmypy"]}, "/bin/dmypy", True, ["/path/to/dmypy"]),
+        ("dmypy", {"dmypy_command": ["/path/to/dmypy"]}, None, True, ["/path/to/dmypy"]),
+        ("mypy", {}, ["/bin/mypy"], False, ["mypy"]),
+        ("mypy", {}, None, False, []),
+        ("mypy", {"mypy_command": ["/path/to/mypy"]}, "/bin/mypy", False, ["mypy"]),
+        ("mypy", {"mypy_command": ["/path/to/mypy"]}, None, False, []),
+        ("dmypy", {}, "/bin/dmypy", False, ["dmypy"]),
+        ("dmypy", {}, None, False, []),
+        ("dmypy", {"dmypy_command": ["/path/to/dmypy"]}, "/bin/dmypy", False, ["dmypy"]),
+        ("dmypy", {"dmypy_command": ["/path/to/dmypy"]}, None, False, []),
+    ],
+)
+def test_get_cmd(command, settings, cmd_on_path, environmentVariableSet: bool, expected):
+    with patch("shutil.which", return_value=cmd_on_path):
+        if environmentVariableSet:
+            os.environ["PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION"] = "Does not matter at all"
+        else:
+            os.environ.pop("PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION", None)
+        assert plugin.get_cmd(settings, command) == expected
+
+
+def test_config_overrides_mypy_command(last_diagnostics_monkeypatch, workspace):
+    last_diagnostics_monkeypatch.setattr(
+        FakeConfig,
+        "plugin_settings",
+        lambda _, p: (
+            {
+                "mypy_command": ["/path/to/mypy"],
+            }
+            if p == "pylsp_mypy"
+            else {}
+        ),
+    )
+
+    m = Mock(wraps=lambda a, **_: Mock(returncode=0, **{"stdout": ""}))
+    last_diagnostics_monkeypatch.setattr(plugin.subprocess, "run", m)
+
+    document = Document(DOC_URI, workspace, DOC_TYPE_ERR)
+
+    config = FakeConfig(uris.to_fs_path(workspace.root_uri))
+    os.environ["PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION"] = "Does not matter at all"
+    plugin.pylsp_settings(config)
+
+    plugin.pylsp_lint(
+        config=config,
+        workspace=workspace,
+        document=document,
+        is_saved=False,
+    )
+
+    called_argv = m.call_args.args[0]
+    called_cmd = called_argv[0]
+    assert called_cmd == "/path/to/mypy"
+
+
+def test_config_overrides_dmypy_command(last_diagnostics_monkeypatch, workspace):
+    last_diagnostics_monkeypatch.setattr(
+        FakeConfig,
+        "plugin_settings",
+        lambda _, p: (
+            {
+                "dmypy": True,
+                "live_mode": False,
+                "dmypy_command": ["poetry", "run", "dmypy"],
+            }
+            if p == "pylsp_mypy"
+            else {}
+        ),
+    )
+
+    m = Mock(wraps=lambda a, **_: Mock(returncode=0, **{"stdout": ""}))
+    last_diagnostics_monkeypatch.setattr(plugin.subprocess, "run", m)
+
+    document = Document(DOC_URI, workspace, DOC_TYPE_ERR)
+
+    config = FakeConfig(uris.to_fs_path(workspace.root_uri))
+    os.environ["PYLSP_MYPY_ALLOW_DANGEROUS_CODE_EXECUTION"] = "Does not matter at all"
+    plugin.pylsp_settings(config)
+
+    plugin.pylsp_lint(
+        config=config,
+        workspace=workspace,
+        document=document,
+        is_saved=False,
+    )
+
+    called_argv = m.call_args.args[0]
+    called_cmd = called_argv[:3]
+    assert called_cmd == ["poetry", "run", "dmypy"]