Uses simple yaml
based rules to take action on JSON
events. Uses jsonpath to scan the event message and regex
for includes
and excludes
conditionals.
Note!
jsonrouter
currently converts all matching field values tostr
ings forre
gex comparision so be aware of this when expectingint
to be returned.
pip install jsonrouter
import json
import yaml
from jsonrouter import JsonMatchEngine
# load rules from file or string
configs = yaml.load('''
rules:
- name: example-rule
routers:
- name: print-router
vars:
- name: the-name
jsonpath: $.name
''')
# an example json record
json_string = '''
{
"name": "jsonrouter",
"type": "jsonpath matcher and router",
"why": {
"because": "it's easy"
}
}
'''
def print_router(data):
# a trivial example router function
print(json.dumps(data, indent=4))
# explicitly declare your registered routers for security
registered_routers = {
'print-router': print_router
}
eng = JsonMatchEngine(configs, registered_routers)
Use the engine to find matches:
In [2]: matches = eng.route_matches(json_string)
{
"name": "example-rule",
"routers": [
{
"name": "print-router"
}
],
"vars": {
"the-name": "jsonrouter"
},
"template": "",
"record": {
"name": "jsonrouter",
"type": "jsonpath matcher and router",
"why": {
"because": "it's easy"
}
}
}
The contents of matches
:
In [3]: matches
Out[3]:
[{'name': 'example-rule',
'routers': [{'name': 'print-router'}],
'vars': {'the-name': 'jsonrouter'},
'template': '',
'record': {'name': 'jsonrouter',
'type': 'jsonpath matcher and router',
'why': {'because': "it's easy"}}}]
Anatomy of the config.
# `rules` is the root of the config
# note: rules can have same name!
rules: # required
- name: notification # required
routers: # required
- name: slack # required
# all fields besides name are optional but may be required in the router
channel: my-channel # optional
vars: # required
- name: type # required
jsonpath: $..Type # required, except for constants: see bellow
includes: ['.*'] # optional, default `['.*']` includes all
excludes: [] # optional, default `[]` excludes nothing
# `template` is optional
template: |
This {type} just came in
You can define a constant var by providing value
field only
vars:
- name: my-constant
value: my constant value
Simple capture examples.
rules:
- name: notification
routers:
- name: slack
channel: my-channel
vars:
- name: type
jsonpath: $..Type
# include anything
includes: ['.*']
# exclude empty
excludes: ['^$']
template: |
This {type} just came in
rules:
- name: console_login
routers:
- name: slack
channel: my-channel
vars:
- name: detail-type
jsonpath: $..detail-type
includes: ['AWS Console Sign In via CloudTrail']
excludes: ['^$']
- name: user
jsonpath: $..principalId
# Match part of string
includes: ['.*:(.*)']
excludes: ['^$']
- name: account
jsonpath: $..account
includes: ['.*']
excludes: ['^$']
template: |
Yo! {user} just signed in to {account}.
Use (?P<variable_name>)
to capture patterns within the matched field.
This will override any naming collisions with
vars:name
you set in the yaml. It merges the rule vars with matched name(s) declared in the regex where named regex take precedence
rules:
- name: console_login
routers:
- name: slack
channel: my-channel
vars:
- name: detail-type
jsonpath: $..detail-type
includes: ['AWS Console Sign In via CloudTrail']
excludes: ['^$']
- name: user
jsonpath: $..principalId
# Match part of string with variable names
includes: ['(?P<stuff>.*):(?P<user>.*)']
excludes: ['^$']
- name: account
jsonpath: $..account
includes: ['.*']
excludes: ['^$']
template: |
Yo! {user} just signed in to {account}. This {stuff} was before the user.
Since jsonrouter
started from an AWS Lambda SNS parsing project it has a jsonify_string
method that converts the SNS message field from a sting to a nested dict
.
import json
import yaml
from jsonrouter import JsonMatchEngine, jsonify_string
def slack(webhook):
# whatever logic you want in here
pass
with open('rules.yaml', 'r') as f:
configs = yaml.safe_load(f)
registered_routers = {
'slack': slack
}
eng = JsonMatchEngine(configs, registered_routers)
def handler(event, context):
# Main lambda handler function
eng.route_matches(jsonify_string(event)['Records'])
# local tests run from root of project
python3 -m pytest -v
# to see print output use `-s`
python3 -m pytest -v -s