1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
import json
import sys
import jsonschema
# fmt:off
# NOTE: Commented out properties are left intentionally, so it is easier to see
# what properties are optional.
schema = {
"$schema": "http://json-schema.org/schema#",
"type": "object",
"properties": {
"document_version": {"type": "integer"},
"ip": {"type": "string"},
"port": {"type": "integer"},
"whois_description": {"type": "string"},
"asn": {"type": "string"},
"asn_country_code": {"type": "string"},
"ptr": {"type": "string"},
"abuse_mail": {"type": "string"},
"domain": {"type": "string"},
"timestamp_in_utc": {"type": "string"},
"display_name": {"type": "string"},
"description": {"type": "string"},
"custom_data": {
"type": "object",
"patternProperties": {
".*": {
"type": "object",
"properties": {
"display_name": {"type": "string"},
"data": {"type": ["string", "boolean", "integer"]},
"description": {"type": "string"},
},
"required": [
"display_name",
"data",
# "description"
]
},
},
},
"result": {
"type": "object",
"patternProperties": {
".*": {
"type": "object",
"properties": {
"display_name": {"type": "string"},
"vulnerable": {"type": "boolean"},
"investigation_needed": {"type": "boolean"},
"reliability": {"type": "integer"},
"description": {"type": "string"},
},
"oneOf": [
{
"required": [
"display_name",
"vulnerable",
# "reliability", # TODO: reliability is required if vulnerable = true
# "description",
]
},
{
"required": [
"display_name",
"investigation_needed",
# "reliability", # TODO: reliability is required if investigation_needed = true
# "description",
]
},
]
},
},
},
},
"required": [
"document_version",
"ip",
"port",
"whois_description",
"asn",
"asn_country_code",
"ptr",
"abuse_mail",
"domain",
"timestamp_in_utc",
"display_name",
# "description",
# "custom_data",
"result",
],
}
# fmt:on
def validate_collector_data(json_blob):
try:
jsonschema.validate(json_blob, schema)
except jsonschema.exceptions.ValidationError as e:
print(f"Validation failed with error: {e}")
return False
return True
if __name__ == "__main__":
with open(sys.argv[1]) as fd:
json_data = json.loads(fd.read())
validate_collector_data(json_data)
|
