ht_test_rule.py

ht_test_rule.py

from python_rules import Rule, deep_get
import rapidjson

def original_get(e, key='event.original', default=None):
    # used to return event.original field, deserialized into dictionary
    # if key not found, we return empty dict
    nested = deep_get(e, *key.split('.'))
    if nested is None:
        return default
    try:
        # deserialize, for example a field with '.flattened.' or 'event.original'
        return rapidjson.loads(nested)
    except:  # couldn't be deseralized
        return nested
        
class HoneyTrapCounterRule(Rule):
    id = "9c2ead36-df11-4b0e-bb8e-8a99234a9d88"
    title = "HoneyTrap test rule"
    description = "Alerts whenever a remote address hits more than five ports"
    author = "Bouke Hendriks"
    date = "2021/04/22"
    tags = []
    status = "experimental"
    level = "medium"

    def rule(self, e):
        client_ip = e.get('client', {}).get('ip')
        client_port = e.get('client', {}).get('port')
        
        count_ports = self.stats.groupby('client.ip').windowed('1d').get('count', 'client.port')
        
        self.description = f"Client IP: {client_ip} has hit {count_ports} ports in a day!"
        
        return count_ports > 5