requests and scrpli_netconf support

requirements/base.txt

@@ -5,6 +5,7 @@ jq>=1.4.0,<2
 deepdiff>=6.2.0,<7
 simpleeval==0.9.*
 netmiko>=4.0.0,<5
+scrapli_netconf==2024.1.30
 textfsm>=1.1.3,<2
 xmltodict<1
 

validity/choices.py

@@ -112,9 +112,11 @@ def pk_field(self):
 
 
 class ConnectionTypeChoices(TextChoices, metaclass=ColoredChoiceMeta):
-    netmiko = "netmiko", "blue"
+    netmiko = "netmiko", "netmiko", "blue"
+    requests = "requests", "requests", "info"
+    scrapli_netconf = "scrapli_netconf", "scrapli_netconf", "orange"
 
-    __command_types__ = {"netmiko": "CLI"}
+    __command_types__ = {"netmiko": "CLI", "scrapli_netconf": "netconf", "requests": "json_api"}
 
     @property
     def acceptable_command_type(self) -> "CommandTypeChoices":
@@ -123,9 +125,18 @@ def acceptable_command_type(self) -> "CommandTypeChoices":
 
 class CommandTypeChoices(TextChoices, metaclass=ColoredChoiceMeta):
     CLI = "CLI", "CLI", "blue"
+    netconf = "netconf", "orange"
+    json_api = "json_api", "JSON API", "info"
 
 
 class ExplanationVerbosityChoices(IntegerChoices):
     disabled = 0, _("0 - Disabled")
     medium = 1, _("1 - Medium")
     maximum = 2, _("2 - Maximum")
+
+
+class JSONAPIMethodChoices(TextChoices):
+    GET = "GET"
+    POST = "POST"
+    PATCH = "PATCH"
+    PUT = "PUT"

validity/forms/general.py

@@ -116,14 +116,23 @@ class Meta:
 
 class PollerForm(NetBoxModelForm):
     commands = DynamicModelMultipleChoiceField(queryset=models.Command.objects.all())
+    public_credentials = CharField(
+        required=False,
+        help_text=_("Enter non-private parameters of the connection type in JSON format."),
+        widget=Textarea(attrs={"style": "font-family:monospace"}),
+    )
+    private_credentials = CharField(
+        required=False,
+        help_text=_(
+            "Enter private parameters of the connection type in JSON format. "
+            "All the values are going to be encrypted."
+        ),
+        widget=Textarea(attrs={"style": "font-family:monospace"}),
+    )
 
     class Meta:
         model = models.Poller
         fields = ("name", "commands", "connection_type", "public_credentials", "private_credentials", "tags")
-        widgets = {
-            "public_credentials": Textarea(attrs={"style": "font-family:monospace"}),
-            "private_credentials": Textarea(attrs={"style": "font-family:monospace"}),
-        }
 
     def clean(self):
         connection_type = self.cleaned_data.get("connection_type") or get_field_value(self, "connection_type")

validity/models/polling.py

@@ -11,7 +11,7 @@
 from validity.fields import EncryptedDictField
 from validity.managers import CommandQS, PollerQS
 from validity.pollers import get_poller
-from validity.subforms import CLICommandForm
+from validity.subforms import CLICommandForm, JSONAPICommandForm, NetconfCommandForm
 from .base import BaseModel, SubformMixin
 from .serializer import Serializer
 
@@ -51,7 +51,7 @@ class Command(SubformMixin, BaseModel):
 
     subform_type_field = "type"
     subform_json_field = "parameters"
-    subforms = {"CLI": CLICommandForm}
+    subforms = {"CLI": CLICommandForm, "json_api": JSONAPICommandForm, "netconf": NetconfCommandForm}
 
     class Meta:
         ordering = ("name",)

validity/pollers/factory.py

@@ -3,6 +3,8 @@
 from validity.choices import ConnectionTypeChoices
 from .base import DevicePoller
 from .cli import NetmikoPoller
+from .http import RequestsPoller
+from .netconf import ScrapliNetconfPoller
 
 
 if TYPE_CHECKING:
@@ -16,7 +18,13 @@ def __init__(self, poller_map: dict) -> None:
     def __call__(self, connection_type: str, credentials: dict, commands: Sequence["Command"]) -> DevicePoller:
         if poller_cls := self.poller_map.get(connection_type):
             return poller_cls(credentials=credentials, commands=commands)
-        raise KeyError("No poller exist for this connection type", connection_type)
+        raise KeyError("No poller exists for this connection type", connection_type)
 
 
-get_poller = PollerFactory(poller_map={ConnectionTypeChoices.netmiko: NetmikoPoller})
+get_poller = PollerFactory(
+    poller_map={
+        ConnectionTypeChoices.netmiko: NetmikoPoller,
+        ConnectionTypeChoices.requests: RequestsPoller,
+        ConnectionTypeChoices.scrapli_netconf: ScrapliNetconfPoller,
+    }
+)

validity/pollers/http.py

@@ -0,0 +1,60 @@
+import json
+from typing import TYPE_CHECKING
+
+import jq
+import requests
+from dcim.models import Device
+from pydantic import BaseModel, Field
+
+from validity.j2_env import Environment
+from validity.utils.misc import process_json_values, reraise
+from .base import ConsecutivePoller
+from .exceptions import PollingError
+
+
+if TYPE_CHECKING:
+    from validity.models import Command, VDevice
+
+
+class RequestParams(BaseModel, extra="allow"):
+    url: str = Field("https://{{device.primary_ip}}/{{command.parameters.url_path.lstrip('/')}}", exclude=True)
+    verify: bool | str = False
+
+    def rendered_url(self, device: "Device", command: "Command") -> str:
+        return Environment().from_string(self.url).render(device=device, command=command)
+
+
+class HttpDriver:
+    def __init__(self, device: Device, **poller_credentials) -> None:
+        self.device = device
+        self.request_params = RequestParams.model_validate(poller_credentials)
+
+    def render_body(self, orig_body: dict, command: "Command"):
+        return process_json_values(
+            orig_body,
+            match_fn=lambda val: isinstance(val, str),
+            transform_fn=lambda val: Environment().from_string(val).render(device=self.device, command=command),
+        )
+
+    def request(self, command: "Command") -> str:
+        request_kwargs = self.request_params.model_dump()
+        request_kwargs["url"] = self.request_params.rendered_url(self.device, command)
+        request_kwargs["method"] = command.parameters["method"]
+        if body := self.render_body(command.parameters["body"], command):
+            request_kwargs["json"] = body
+        return requests.request(**request_kwargs).content.decode()
+
+
+class RequestsPoller(ConsecutivePoller):
+    driver_cls = HttpDriver
+
+    def get_credentials(self, device: "VDevice"):
+        return self.credentials | {"device": device}
+
+    def poll_one_command(self, driver: HttpDriver, command: "Command") -> str:
+        answer = driver.request(command)
+        with reraise((json.JSONDecodeError, ValueError), PollingError, device_wide=False):
+            json_answer = json.loads(answer)
+            if jq_query := command.parameters.get("jq_query"):
+                json_answer = jq.first(jq_query, json_answer)
+            return json.dumps(json_answer)

validity/pollers/netconf.py

@@ -0,0 +1,18 @@
+from typing import TYPE_CHECKING
+
+from scrapli_netconf.driver import NetconfDriver
+
+from .base import ConsecutivePoller
+
+
+if TYPE_CHECKING:
+    from validity.models import Command
+
+
+class ScrapliNetconfPoller(ConsecutivePoller):
+    driver_cls = NetconfDriver
+    host_param_name = "host"
+
+    def poll_one_command(self, driver: NetconfDriver, command: "Command") -> str:
+        response = driver.rpc(command.parameters["rpc"])
+        return response.result

validity/subforms.py

@@ -3,10 +3,38 @@
     1. Render part of the main form for JSON Field
     2. Validate JSON Field
 """
+import xml.etree.ElementTree as ET
+
 from django import forms
 from django.utils.translation import gettext_lazy as _
 from utilities.forms import BootstrapMixin
 
+from validity.choices import JSONAPIMethodChoices
+from validity.utils.misc import reraise
+
 
 class CLICommandForm(BootstrapMixin, forms.Form):
     cli_command = forms.CharField(label=_("CLI Command"))
+
+
+class JSONAPICommandForm(BootstrapMixin, forms.Form):
+    method = forms.ChoiceField(label=_("Method"), initial="GET", choices=JSONAPIMethodChoices.choices)
+    url_path = forms.CharField(label=_("URL Path"))
+    jq_query = forms.CharField(
+        label=_("JQ Query"), required=False, help_text=_("Process API answer with this JQ expression")
+    )
+    body = forms.JSONField(
+        label=_("Body"),
+        required=False,
+        help_text=_("Enter data in JSON format. You can use Jinja2 expressions as values."),
+    )
+
+
+class NetconfCommandForm(BootstrapMixin, forms.Form):
+    rpc = forms.CharField(label=_("RPC"), widget=forms.Textarea(attrs={"placeholder": "<get-config/>"}))
+
+    def clean_rpc(self):
+        rpc = self.cleaned_data["rpc"]
+        with reraise(Exception, forms.ValidationError, {"rpc": "Invalid XML"}):
+            ET.fromstring(rpc)
+        return rpc

validity/utils/misc.py

@@ -1,8 +1,9 @@
+import copy
 import inspect
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager, suppress
 from itertools import islice
-from typing import TYPE_CHECKING, Any, Callable, Iterable
+from typing import TYPE_CHECKING, Any, Callable, Collection, Iterable
 
 from core.exceptions import SyncError
 from django.utils.html import format_html
@@ -91,3 +92,26 @@ def batched(iterable: Iterable, n: int, container: type = list):
         if not batch:
             return
         yield batch
+
+
+Json = dict[str, "Json"] | list["Json"] | int | float | str | None
+
+
+def process_json_values(data: Json, match_fn: Callable[[Json], bool], transform_fn: Callable[[Json], Json]) -> Json:
+    """
+    Traverse JSON-like struct recursively and apply "tranform_fn" to values matched by "match_fn"
+    """
+
+    def transform(data_item: Json) -> None:
+        if not isinstance(data_item, Collection):
+            return
+        iterator = data_item.items() if isinstance(data_item, dict) else enumerate(data_item)
+        for key, value in iterator:
+            if match_fn(value):
+                data_item[key] = transform_fn(value)
+            elif isinstance(value, Collection):
+                transform(value)
+
+    data_copy = copy.deepcopy(data)
+    transform(data_copy)
+    return data_copy