add view for web hook test and add recursion to the dictionary data r…

…eplacer

api/tacticalrmm/core/urls.py

@@ -16,6 +16,7 @@
  path("urlaction/", views.GetAddURLAction.as_view()),
  path("urlaction/<int:pk>/", views.UpdateDeleteURLAction.as_view()),
  path("urlaction/run/", views.RunURLAction.as_view()),
+ path("urlaction/run/test/", views.RunTestURLAction.as_view()),
  path("servertasks/", views.GetAddServerTasks.as_view()),
  path("servertasks/<int:pk>/", views.UpdateDeleteServerTask.as_view()),
  path("servertasks/<int:pk>/run/", views.RunServerTask.as_view()),

api/tacticalrmm/core/utils.py

@@ -14,6 +14,7 @@
 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
 
 from tacticalrmm.constants import (
@@ -227,17 +228,36 @@ def find_and_replace_db_values_str(*, text: str, instance):
  return text
 
 
-def find_and_replace_db_values_dict(*, dict: Dict[str, Any], instance):
+def find_and_replace_db_values_dict(*, dict_value: Dict[str, Any], instance)-> Dict[str, Any]:
  new_dict = {}
 
- for key, value in dict.items():
+ for key, value in dict_value.items():
  new_key = find_and_replace_db_values_str(text=key, instance=instance)
- new_value = find_and_replace_db_values_str(text=value, instance=instance)
+
+ # Check if the value is a dictionary and recursively call the function if it is
+ if isinstance(value, dict):
+ new_value = find_and_replace_db_values_dict(dict_value=value, instance=instance)
+ else:
+ new_value = find_and_replace_db_values_str(text=value, instance=instance)
+
  new_dict[new_key] = new_value
 
  return new_dict
 
 
+def _run_url_rest_action(*, url: str, method, body: str, headers: str, instance=None):
+ # replace url
+ new_url = find_and_replace_db_values_str(text=url, instance=instance)
+ new_body = find_and_replace_db_values_dict(dict_value=json.loads(body), instance=instance)
+ new_headers = find_and_replace_db_values_dict(
+ dict_value=json.loads(headers), instance=instance
+ )
+
+ if method in ["get", "delete"]:
+ return getattr(requests, method)(new_url, headers=new_headers)
+ else:
+ return getattr(requests, method)(new_url, data=new_body, headers=new_headers)
+
 def run_url_rest_action(*, action_id: int, instance=None) -> Tuple[str, int]:
  import core.models
 
@@ -247,20 +267,26 @@ def run_url_rest_action(*, action_id: int, instance=None) -> Tuple[str, int]:
  body = action.rest_body
  headers = action.rest_headers
 
- # replace url
- url = find_and_replace_db_values_str(text=url, instance=instance)
- body = find_and_replace_db_values_dict(dict=json.loads(body), instance=instance)
- headers = find_and_replace_db_values_dict(
- dict=json.loads(headers), instance=instance
- )
-
- if method in ["get", "delete"]:
- response = getattr(requests, method)(url, headers=headers)
- else:
- response = getattr(requests, method)(url, data=body, headers=headers)
+ response = _run_url_rest_action(url=url, method=method, body=body, headers=headers, instance=instance)
 
  return (response.text, response.status_code)
 
+def run_test_url_rest_action(*, url: str, method, body: str, headers: str, instance_type: Optional[str], instance_id: Optional[int]) -> Tuple[str, int]:
+ lookup_instance = None
+ if instance_type and instance_id:
+ if instance_type == "client":
+ Client = apps.get_model("clients.Client")
+ lookup_instance = Client.objects.get(pk=instance_id)
+ elif instance_type == "site":
+ Site = apps.get_model("clients.Site")
+ lookup_instance = Site.objects.get(pk=instance_id)
+ elif instance_type == "agent":
+ Agent = apps.get_model("agents.Agent")
+ lookup_instance = Agent.objects.get(pk=instance_id) 
+
+ response = _run_url_rest_action(url=url, method=method, body=body, headers=headers, instance=lookup_instance)
+
+ return (response.text, response.status_code)
 
 def run_server_task(*, server_task_id: int):
  from autotasks.models import AutomatedTask, TaskResult

api/tacticalrmm/core/views.py

@@ -29,6 +29,7 @@
  token_is_valid,
  run_server_script,
  run_server_task,
+ run_test_url_rest_action
 )
 from logs.models import AuditLog
 from tacticalrmm.constants import AuditActionType, PAStatus
@@ -423,6 +424,31 @@ def patch(self, request):
 
  return Response(requote_uri(url_pattern))
 
+class RunTestURLAction(APIView):
+ permission_classes = [IsAuthenticated, URLActionPerms]
+
+ class InputSerializer(serializers.Serializer):
+ pattern = serializers.CharField(required=True)
+ rest_body = serializers.CharField()
+ rest_headers = serializers.CharField()
+ rest_method = serializers.ChoiceField(required=True, choices=["get", "post", "put", "delete", "patch"])
+ run_instance_type = serializers.ChoiceField(choices=["agent", "client", "site", "none"])
+ run_instance_id = serializers.IntegerField(allow_null=True)
+
+ def post(self, request):
+
+ serializer = self.InputSerializer(data=request.data)
+ serializer.is_valid(raise_exception=True)
+
+ url = serializer.validated_data.get("pattern")
+ body = serializer.validated_data.get("rest_body", None)
+ headers = serializer.validated_data.get("rest_headers", None)
+ method = serializer.validated_data.get("rest_method")
+ instance_type = serializer.validated_data.get("run_instance_type", None)
+ instance_id = serializer.validated_data.get("instance_id", None)
+ result, _ = run_test_url_rest_action(url=url, body=body, headers=headers, method=method, instance_type=instance_type, instance_id=instance_id)
+
+ return Response({"url": url, "result": result})
 
 class ServerTaskSerializer(serializers.ModelSerializer):