We’re excited to introduce support for custom validators in Schema Enforcer. Schema Enforcer provides a framework for testing structured data against schema definitions using JSON Schema and now using custom Python validators. You can check out Introducing Schema Enforcer for more background and an introduction to Schema Enforcer.

What Is a Custom Validator?

A custom validator is a Python module that allows you to run any logic against your data on a per-host basis.

Let’s start with an example. What if you want to validate that every edge router has at least two core interfaces defined?

Here’s a possible way we could model our data in an Ansible host_var file:

---
hostname: "az-phx-pe01"
pair_rtr: "az-phx-pe02"
upstreams: []
interfaces:
  MgmtEth0/0/CPU0/0:
    ipv4: "172.16.1.1"
  Loopback0:
    ipv4: "192.168.1.1"
    ipv6: "2001:db8:1::1"
  GigabitEthernet0/0/0/0:
    ipv4: "10.1.0.1"
    ipv6: "2001:db8::"
    peer: "az-phx-pe02"
    peer_int: "GigabitEthernet0/0/0/0"
    type: "core"
  GigabitEthernet0/0/0/1:
    ipv4: "10.1.0.37"
    ipv6: "2001:db8::12"
    peer: "co-den-p01"
    peer_int: "GigabitEthernet0/0/0/2"
    type: "core"

In this example, each physical interface has a type key, which we can evaluate in our custom validator. JSON Schema can be used to validate that this field exists and contains a desired value (e.g., “core”, “access”, etc.). However, it cannot check whether there are at least two interfaces with this key set to “core”.

JMESPath Custom Validators

As a shortcut for basic use cases, Schema Enforcer provides the JmesPathModelValidation class. This class supports using JMESPath queries against your data along with generic comparison operators. The logic is provided by the base class, so no Python is required beyond setting a few variables.

To solve the preceding example, we can use the following custom validator:

from schema_enforcer.schemas.validator import JmesPathModelValidation

class CheckInterface(JmesPathModelValidation):  # pylint: disable=too-few-public-methods
    top_level_properties = ["interfaces"]
    id = "CheckInterface"  # pylint: disable=invalid-name
    left = "interfaces.*[@.type=='core'][] | length([?@])"
    right = 2
    operator = "gte"
    error = "Less than two core interfaces"

The top_level_properties variable maps this validator to the interfaces object in our data. The real work is done by the left, right, and operator variables. Think of these as part of an expression:

{left} {operator} {right}

Or for our example:

"interfaces.*[@.type=='core'][] | length([?@])" gte 2

This custom validator uses the JMESPath expression to query the data. The query returns all interfaces that have type of “core”. The output is piped to a built-in JMESPath function that gives us the length of the return value. When applied to our example data, the value of the query is 2. When checked by our custom validator, this host will pass, as the value of the query is greater than or equal to 2.

root@b295daf33db5:/local/examples/ansible3# schema-enforcer ansible --show-checks
Found 2 hosts in the inventory
Ansible Host              Schema ID
--------------------------------------------------------------------------------
az_phx_pe01               ['CheckInterface']
az_phx_pe02               ['CheckInterface']

In the preceding output, we see the CheckInterface validator is applied to two hosts.

When Schema Enforcer is run against the inventory, the output shows if any hosts fail the validation. If a host fails, the error message defined in the CheckInterface class error variable will be shown.

root@b295daf33db5:/local/examples/ansible3# schema-enforcer ansible
Found 2 hosts in the inventory
FAIL | [ERROR] Less than two core interfaces [HOST] az_phx_pe02 [PROPERTY]
root@b295daf33db5:/local/examples/ansible3#

Advanced Use Cases

For more advanced use cases, Schema Enforcer provides the BaseValidation class which can be used to build your own complex validation classes. BaseValidation provides two helper functions for reporting pass/fail: add_validation_pass and add_validation_error. Schema Enforcer will automatically call the validate method of your custom class for all instances of your data. The logic as to whether a validator passes or fails is up to your implementation.

Since we can run arbitrary logic against the data using Python, one possible use case is to check data against some external service. In the example below, a simple BGP peer data file is checked against the ARIN database to validate that the name is correct.

Sample Data

---
bgp_peers:
  - asn: 6939
    name: "Hurricane Electric LLC"
  - asn: 701
    name: "VZW"
  - asn: 100000
    name: "Private"

Validator

"""Custom validator for BGP peer information."""
import requests

from schema_enforcer.schemas.validator import BaseValidation


class CheckARIN(BaseValidation):
    """Verify that BGP peer name matches ARIN ASN information."""

    def validate(self, data, strict):
        """Validate BGP peers for each host."""
        headers = {"Accept": "application/json"}
        for peer in data["bgp_peers"]:
            # pylint: disable=invalid-name
            r = requests.get(f"http://whois.arin.net/rest/asn/{peer['asn']}", headers=headers)
            if r.status_code != requests.codes.ok:  # pylint: disable=no-member
                self.add_validation_error(f"ARIN lookup failed for peer {peer['name']} with ASN {peer['asn']}")
                continue
            arin_info = r.json()
            arin_name = arin_info["asn"]["orgRef"]["@name"]
            if peer["name"] != arin_name:
                self.add_validation_error(
                    f"Peer name {peer['name']} for ASN {peer['asn']} does not match ARIN database: {arin_name}"
                )
            else:
                self.add_validation_pass()

If we run Schema Enforcer with this validator, we get the following output:

root@da72aae39ede:/local/examples/example4# schema-enforcer validate --show-checks
Structured Data File                               Schema ID
--------------------------------------------------------------------------------
./bgp/peers.yml                                    ['CheckARIN']
root@da72aae39ede:/local/examples/example4# schema-enforcer validate
FAIL | [ERROR] Peer name VZW for ASN 701 does not match ARIN database: MCI Communications Services, Inc. d/b/a Verizon Business [FILE] ./bgp/peers.yml [PROPERTY]
FAIL | [ERROR] ARIN lookup failed for peer Private with ASN 100000 [FILE] ./bgp/peers.yml [PROPERTY]

You could expand this example to do other validation, such as checking that the ASN is valid before making the request to ARIN.

For more information on this Schema Enforcer feature, see the docs. And if you have any interesting use cases, please let us know!