add test server script and start making server scripts/webterm optional

api/tacticalrmm/alerts/models.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import re
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
 
 from django.contrib.postgres.fields import ArrayField
@@ -105,8 +104,8 @@ def get_result(self):
             return self.assigned_check.checkresults.get(agent=self.agent)
         elif self.alert_type == AlertType.TASK:
             return self.assigned_task.taskresults.get(agent=self.agent)
-        else:
-            return None
+
+        return None
 
     def resolve(self) -> None:
         self.resolved = True
@@ -490,10 +489,11 @@ def handle_alert_failure(
                 and run_script_action
             ):
                 stdout, stderr, execution_time, retcode = run_server_script(
-                    script_id=alert_template.action.pk,
+                    body=alert_template.action.script_body,
                     args=alert.parse_script_args(alert_template.action_args),
                     timeout=alert_template.action_timeout,
                     env_vars=alert.parse_script_args(alert_template.action_env_vars),
+                    shell=alert_template.action.shell,
                 )
 
                 r = {
@@ -658,12 +658,13 @@ def handle_alert_resolve(
                 and run_script_action
             ):
                 stdout, stderr, execution_time, retcode = run_server_script(
-                    script_id=alert_template.resolved_action.pk,
+                    body=alert_template.resolved_action.script_body,
                     args=alert.parse_script_args(alert_template.resolved_action_args),
                     timeout=alert_template.resolved_action_timeout,
                     env_vars=alert.parse_script_args(
                         alert_template.resolved_action_env_vars
                     ),
+                    shell=alert_template.resolved_action.shell,
                 )
                 r = {
                     "stdout": stdout,
@@ -719,7 +720,7 @@ def parse_script_args(self, args: List[str]) -> List[str]:
 
         for arg in args:
             temp_arg = arg
-            for string, model, prop in re.findall(RE_DB_VALUE, arg):
+            for string, model, prop in RE_DB_VALUE.findall(arg):
                 value = get_db_value(string=f"{model}.{prop}", instance=self)
 
                 if value is not None:

api/tacticalrmm/core/consumers.py

@@ -18,6 +18,7 @@
 from django.utils import timezone as djangotime
 
 from agents.models import Agent
+from core.models import CoreSettings
 from tacticalrmm.constants import AgentMonType
 from tacticalrmm.helpers import days_until_cert_expires
 from tacticalrmm.logger import logger
@@ -157,6 +158,11 @@ def connect(self):
             self.close(4401)
             return
 
+        core: CoreSettings = CoreSettings.objects.first()  # type: ignore
+        if not core.web_terminal_enabled:
+            self.close(4401)
+            return
+
         if self.user.block_dashboard_login or not _has_perm(
             self.user, "can_run_servercli"
         ):

api/tacticalrmm/core/migrations/0046_coresettings_enable_server_scripts_and_more.py

@@ -0,0 +1,23 @@
+# Generated by Django 4.2.11 on 2024-04-27 09:50
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0045_urlaction_action_type_urlaction_rest_body_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="coresettings",
+            name="enable_server_scripts",
+            field=models.BooleanField(default=True),
+        ),
+        migrations.AddField(
+            model_name="coresettings",
+            name="enable_server_webterminal",
+            field=models.BooleanField(default=True),
+        ),
+    ]

api/tacticalrmm/core/models.py

@@ -105,6 +105,8 @@ class CoreSettings(BaseAuditModel):
     open_ai_model = models.CharField(
         max_length=255, blank=True, default="gpt-3.5-turbo"
     )
+    enable_server_scripts = models.BooleanField(default=True)
+    enable_server_webterminal = models.BooleanField(default=True)
 
     def save(self, *args, **kwargs) -> None:
         from alerts.tasks import cache_agents_alert_template
@@ -185,6 +187,24 @@ def email_is_configured(self) -> bool:
 
         return False
 
+    @property
+    def server_scripts_enabled(self) -> bool:
+        if getattr(settings, "HOSTED", False) or getattr(
+            settings, "TRMM_DISABLE_SERVER_SCRIPTS", False
+        ):
+            return False
+
+        return self.enable_server_scripts
+
+    @property
+    def web_terminal_enabled(self) -> bool:
+        if getattr(settings, "HOSTED", False) or getattr(
+            settings, "TRMM_DISABLE_WEB_TERMINAL", False
+        ):
+            return False
+
+        return self.enable_server_webterminal
+
     def send_mail(
         self,
         subject: str,

api/tacticalrmm/core/permissions.py

@@ -41,3 +41,8 @@ def has_permission(self, r, view) -> bool:
 class RunServerTaskPerms(permissions.BasePermission):
     def has_permission(self, r, view) -> bool:
         return _has_perm(r, "can_run_servertasks")
+
+
+class WebTerminalPerms(permissions.BasePermission):
+    def has_permission(self, r, view) -> bool:
+        return _has_perm(r, "can_run_servercli")

api/tacticalrmm/core/urls.py

@@ -1,4 +1,5 @@
 from django.urls import path
+from django.conf import settings
 
 from . import views
 
@@ -17,9 +18,18 @@
     path("urlaction/<int:pk>/", views.UpdateDeleteURLAction.as_view()),
     path("urlaction/run/", views.RunURLAction.as_view()),
     path("urlaction/run/test/", views.RunTestURLAction.as_view()),
-    path("serverscript/<int:pk>/run/", views.RunServerScript.as_view()),
     path("smstest/", views.TwilioSMSTest.as_view()),
     path("clearcache/", views.clear_cache),
     path("status/", views.status),
     path("openai/generate/", views.OpenAICodeCompletion.as_view()),
+    path("webtermperms/", views.webterm_perms),
 ]
+
+
+if not (
+    getattr(settings, "HOSTED", False)
+    or getattr(settings, "TRMM_DISABLE_SERVER_SCRIPTS", False)
+):
+    urlpatterns += [
+        path("serverscript/test/", views.TestRunServerScript.as_view()),
+    ]

api/tacticalrmm/core/utils.py

@@ -7,14 +7,13 @@
 from base64 import b64encode
 from contextlib import suppress
 from requests.utils import requote_uri
-from typing import TYPE_CHECKING, List, Optional, cast, Tuple
+from typing import TYPE_CHECKING, Optional, cast, Tuple
 
 import requests
 import websockets
 from django.conf import settings
 from django.core.cache import cache
 from django.http import FileResponse
-from django.utils import timezone as djangotime
 from django.apps import apps
 from meshctrl.utils import get_auth_token
 
@@ -218,15 +217,14 @@ def make_alpha_numeric(s: str):
 
 
 def find_and_replace_db_values_str(*, text: str, instance):
-    import re
     from tacticalrmm.utils import RE_DB_VALUE, get_db_value
 
     if not instance:
         return text
 
     return_string = text
 
-    for string, model, prop in re.findall(RE_DB_VALUE, text):
+    for string, model, prop in RE_DB_VALUE.findall(text):
         value = get_db_value(string=f"{model}.{prop}", instance=instance)
         return_string = return_string.replace(string, str(value))
     return return_string
@@ -241,7 +239,7 @@ def _run_url_rest_action(*, url: str, method, body: str, headers: str, instance=
     new_body = json.loads(new_body)
     new_headers = json.loads(new_headers)
 
-    if method in ["get", "delete"]:
+    if method in ("get", "delete"):
         return getattr(requests, method)(new_url, headers=new_headers)
 
     return getattr(requests, method)(
@@ -282,7 +280,7 @@ def run_test_url_rest_action(
     instance_id: Optional[int],
 ) -> Tuple[str, str, str]:
     lookup_instance = None
-    if instance_type and instance_type in lookup_apps.keys() and instance_id:
+    if instance_type and instance_type in lookup_apps and instance_id:
         app, model = lookup_apps[instance_type]
         Model = apps.get_model(app, model)
         if instance_type == "agent":
@@ -297,91 +295,22 @@ def run_test_url_rest_action(
     return (response.text, response.request.url, response.request.body)
 
 
-def run_server_task(*, server_task_id: int):
-    from autotasks.models import AutomatedTask, TaskResult
-    from tacticalrmm.constants import TaskStatus
-
-    task = AutomatedTask.objects.get(pk=server_task_id)
-    output = ""
-    total_execution_time = 0
-    passing = True
-
-    for action in task.actions:
-        if action["type"] == "cmd":
-            stdout, stderr, execution_time, retcode = run_server_command(
-                command=action["command"], timeout=action["timeout"]
-            )
-            name = action["command"]
-        else:
-            stdout, stderr, execution_time, retcode = run_server_script(
-                script_id=action["script"],
-                args=action["script_args"],
-                env_vars=action["env_vars"],
-                timeout=action["timeout"],
-            )
-            name = action["name"]
-
-        if retcode != 0:
-            passing = False
-
-        total_execution_time += execution_time
-        output += f"-----------------------------\n{name}\n-----------------------------\n\nStandard Output: {stdout}\n\nStandard Error: {stderr}\n"
-        output += f"Return Code: {retcode}\n"
-        output += f"Execution Time: {execution_time}\n\n"
-
-    try:
-        task_result = TaskResult.objects.get(task=task, agent=None)
-        task_result.retcode = 0 if passing else 1
-        task_result.execution_time = total_execution_time
-        task_result.stdout = output
-        task_result.stderr = ""
-        task_result.status = TaskStatus.PASSING if passing else TaskStatus.FAILING
-        task_result.save()
-    except TaskResult.DoesNotExist:
-        TaskResult.objects.create(
-            task=task,
-            agent=None,
-            retcode=0 if passing else 1,
-            execution_time=total_execution_time,
-            stdout=output,
-            stderr="",
-            status=TaskStatus.PASSING if passing else TaskStatus.FAILING,
-            last_run=djangotime.now(),
-        )
-
-    return (output, total_execution_time)
-
-
-def run_server_command(*, command: List[str], timeout: int):
-    start_time = time.time()
-    result = subprocess.run(
-        command, capture_output=True, text=True, shell=True, timeout=timeout
-    )
-    execution_time = time.time() - start_time
-
-    return (result.stdout, result.stderr, execution_time, result.returncode)
-
-
 def run_server_script(
-    *, script_id: int, args: List[str], env_vars: List[str], timeout: int
+    *, body: str, args: list[str], env_vars: list[str], shell: str, timeout: int
 ):
     from scripts.models import Script
 
-    script = Script.objects.get(pk=script_id)
+    parsed_args = Script.parse_script_args(None, shell, args)
 
-    parsed_args = script.parse_script_args(None, script.shell, args)
-
-    parsed_env_vars = script.parse_script_env_vars(
-        None, shell=script.shell, env_vars=env_vars
-    )
+    parsed_env_vars = Script.parse_script_env_vars(None, shell=shell, env_vars=env_vars)
 
     custom_env = os.environ.copy()
     for var in parsed_env_vars:
         var_split = var.split("=")
         custom_env[var_split[0]] = var_split[1]
 
     with tempfile.NamedTemporaryFile(mode="w", delete=False) as tmp_script:
-        tmp_script.write(script.script_body.replace("\r\n", "\n"))
+        tmp_script.write(body.replace("\r\n", "\n"))
         tmp_script_path = tmp_script.name
 
     os.chmod(tmp_script_path, 0o550)