Merge pull request #409

CHANGES

@@ -1,3 +1,7 @@
+2.8.0-dev.80 | 2024-05-31 09:10:03 +0200
+
+  * Add JSON schema for the WebSocket output format (Dominik Charousset, Corelight)
+
 2.8.0-dev.78 | 2024-05-31 09:09:10 +0200
 
   * Remove workaround for unsupported CMake versions (Dominik Charousset, Corelight)

VERSION

@@ -1 +1 @@
-2.8.0-dev.78
+2.8.0-dev.80

api/v1-broker-out.json

@@ -0,0 +1,251 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "Messages that Broker Sends over WebSocket",
+  "type": "object",
+  "anyOf": [
+    { "$ref": "#/definitions/AckMessage" },
+    { "$ref": "#/definitions/DataMessage" },
+    { "$ref": "#/definitions/ErrorMessage" }
+  ],
+  "definitions": {
+    "AckMessage": {
+      "properties": {
+        "type": {
+          "const": "ack"
+        },
+        "endpoint": {
+          "type": "string"
+        },
+        "version": {
+          "type": "string"
+        }
+      },
+      "required": ["type", "endpoint", "version"]
+    },
+    "DataMessage": {
+      "$ref": "#/definitions/Data",
+      "properties": {
+        "type": {
+          "const": "data-message"
+        },
+        "topic": {
+          "type": "string"
+        }
+      },
+      "required": ["type", "topic"]
+    },
+    "ErrorMessage": {
+      "properties": {
+        "type": {
+          "const": "error"
+        },
+        "code": {
+          "type": "string"
+        },
+        "context": {
+          "type": "string"
+        }
+      },
+      "required": ["type", "code", "context"]
+    },
+    "Data": {
+      "anyOf": [
+        { "$ref": "#/definitions/None" },
+        { "$ref": "#/definitions/Boolean" },
+        { "$ref": "#/definitions/Count" },
+        { "$ref": "#/definitions/Integer" },
+        { "$ref": "#/definitions/Real" },
+        { "$ref": "#/definitions/Timespan" },
+        { "$ref": "#/definitions/Timestamp" },
+        { "$ref": "#/definitions/String" },
+        { "$ref": "#/definitions/EnumValue" },
+        { "$ref": "#/definitions/Address" },
+        { "$ref": "#/definitions/Subnet" },
+        { "$ref": "#/definitions/Port" },
+        { "$ref": "#/definitions/Vector" },
+        { "$ref": "#/definitions/Set" },
+        { "$ref": "#/definitions/Table" }
+      ]
+    },
+    "None": {
+      "properties": {
+        "@data-type": {
+          "const": "none"
+        },
+        "data": { }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Boolean": {
+      "properties": {
+        "@data-type": {
+          "const": "boolean"
+        },
+        "data": {
+          "type": "boolean"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Count": {
+      "properties": {
+        "@data-type": {
+          "const": "count"
+        },
+        "data": {
+          "type": "integer",
+          "minimum" : 0
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Integer": {
+      "properties": {
+        "@data-type": {
+          "const": "integer"
+        },
+        "data": {
+          "type": "integer"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Real": {
+      "properties": {
+        "@data-type": {
+          "const": "real"
+        },
+        "data": {
+          "type": "number"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Timespan": {
+      "properties": {
+        "@data-type": {
+          "const": "timespan"
+        },
+        "data": {
+          "type": "string",
+          "pattern": "^[0-9]+(ns|ms|s|min|h|d)$"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Timestamp": {
+      "properties": {
+        "@data-type": {
+          "const": "timestamp"
+        },
+        "data": {
+          "type": "string",
+          "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\\.[0-9]+)?$"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "String": {
+      "properties": {
+        "@data-type": {
+          "const": "string"
+        },
+        "data": {
+          "type": "string"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "EnumValue": {
+      "properties": {
+        "@data-type": {
+          "const": "enum-value"
+        },
+        "data": {
+          "type": "string"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Address": {
+      "properties": {
+        "@data-type": {
+          "const": "address"
+        },
+        "data": {
+          "type": "string",
+          "pattern": "^([0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+)|([0-9a-fA-F:.]+)?$"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Subnet": {
+      "properties": {
+        "@data-type": {
+          "const": "subnet"
+        },
+        "data": {
+          "type": "string",
+          "pattern": "^([0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+)|([0-9a-fA-F:.]+)?/[0-9]+$"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Port": {
+      "properties": {
+        "@data-type": {
+          "const": "port"
+        },
+        "data": {
+          "type": "string",
+          "pattern": "^[0-9]+/(tcp|udp|icmp|\\?)?$"
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Vector": {
+      "properties": {
+        "@data-type": {
+          "const": "vector"
+        },
+        "data": {
+          "type": "array",
+          "items": { "$ref": "#/definitions/Data" }
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "Set": {
+      "properties": {
+        "@data-type": {
+          "const": "set"
+        },
+        "data": {
+          "type": "array",
+          "uniqueItems": true,
+          "items": { "$ref": "#/definitions/Data" }
+        }
+      },
+      "required": ["@data-type", "data"]
+    },
+    "KeyValuePair": {
+      "properties": {
+        "key": { "$ref": "#/definitions/Data" },
+        "value": { "$ref": "#/definitions/Data" }
+      },
+      "required": ["key", "value"]
+    },
+    "Table": {
+      "properties": {
+        "@data-type": {
+          "const": "table"
+        },
+        "data": {
+          "type": "array",
+          "items": { "$ref": "#/definitions/KeyValuePair" }
+        }
+      },
+      "required": ["@data-type", "data"]
+    }
+  }
+}

ci/test.sh

@@ -39,7 +39,7 @@ if command -v pip3 >/dev/null 2>&1 ; then
     export PATH="$PATH:$BinDir"
     python3 -m venv test-env
     source test-env/bin/activate
-    pip3 install btest websockets
+    pip3 install btest websockets jsonschema
     cd $BaseDir/tests/btest
     btest || result=1
     [[ -d .tmp ]] && tar -czf tmp.tar.gz .tmp

tests/btest/btest.cfg

@@ -10,5 +10,6 @@ SCRIPTS    = %(testbase)s/scripts
 TMPDIR     = %(testbase)s/.tmp
 TEST_ROOT  = %(testbase)s
 CERTS_ROOT = %(testbase)s/../data/certs
+API_ROOT   = %(testbase)s/../../api
 PYTHONUNBUFFERED=1
 

tests/btest/web-socket/encrypted.py

@@ -7,6 +7,7 @@
 # @TEST-COPY-FILE: ${CERTS_ROOT}/key.1.enc.pem
 # @TEST-COPY-FILE: ${CERTS_ROOT}/cert.2.pem
 # @TEST-COPY-FILE: ${CERTS_ROOT}/key.2.pem
+# @TEST-COPY-FILE: ${API_ROOT}/v1-broker-out.json
 #
 # @TEST-EXEC: btest-bg-run node "broker-node --config-file=../node.cfg"
 # @TEST-EXEC: btest-bg-run recv "python3 ../recv.py >recv.out"
@@ -38,6 +39,8 @@
 
 import asyncio, websockets, os, time, json, sys, ssl
 
+from jsonschema import validate
+
 ws_port = os.environ['BROKER_WEB_SOCKET_PORT'].split('/')[0]
 
 ws_url = f'wss://localhost:{ws_port}/v1/messages/json'
@@ -46,6 +49,8 @@
 ssl_ctx.load_verify_locations("../ca.pem")
 ssl_ctx.load_cert_chain(certfile="../cert.2.pem", keyfile="../key.2.pem")
 
+schema = json.load(open('../v1-broker-out.json'))
+
 async def do_run():
     # Try up to 30 times.
     connected  = False
@@ -56,17 +61,22 @@ async def do_run():
             # send filter and wait for ack
             await ws.send('["/test"]')
             ack_json = await ws.recv()
-            ack = json.loads(ack_json)
-            if not 'type' in ack or ack['type'] != 'ack':
-                print('*** unexpected ACK from server:')
-                print(ack_json)
-                sys.exit()
+            try:
+                validate(json.loads(ack_json), schema)
+            except Exception as err:
+                print(f'received invalid ack: {err}')
+                sys.exit(1)
             # tell btest to start the sender now
             with open('ready', 'w') as f:
                 f.write('ready')
             # dump messages to stdout (redirected to recv.out)
             for i in range(10):
                 msg = await ws.recv()
+                try:
+                    validate(json.loads(msg), schema)
+                except Exception as err:
+                    print(f'received invalid data message: {err}')
+                    sys.exit(1)
                 print(f'{msg}')
             # tell btest we're done
             with open('done', 'w') as f:

tests/btest/web-socket/error.py

@@ -2,6 +2,8 @@
 #
 # @TEST-PORT: BROKER_WEB_SOCKET_PORT
 #
+# @TEST-COPY-FILE: ${API_ROOT}/v1-broker-out.json
+#
 # @TEST-EXEC: btest-bg-run node "broker-node --config-file=../node.cfg"
 # @TEST-EXEC: btest-bg-run send "python3 ../send.py >send.out"
 #
@@ -24,6 +26,8 @@
 
 import asyncio, websockets, os, json, sys, traceback
 
+from jsonschema import validate
+
 ws_port = os.environ['BROKER_WEB_SOCKET_PORT'].split('/')[0]
 
 ws_url = f'ws://localhost:{ws_port}/v1/messages/json'
@@ -34,6 +38,8 @@
     "data": '555-0123'
 }
 
+schema = json.load(open('../v1-broker-out.json'))
+
 async def do_run():
     # Try up to 30 times.
     connected  = False
@@ -48,6 +54,11 @@ async def do_run():
             # are likely to change in the future
             err_json = await ws.recv()
             err = json.loads(err_json)
+            try:
+                validate(err, schema)
+            except Exception as ex:
+                print(f'received invalid error message: {ex}')
+                sys.exit(1)
             want = 'input #1 contained invalid data'
             got = err['context']
             if got.startswith(want):