#!/usr/bin/env python3 # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # Background worker - finds bans and adds 'em, and such and things import asyncio import elasticsearch_dsl import elasticsearch import typing import netaddr import time import plugins.configuration import plugins.lists import datetime MAX_DB_DAYS = 3 # Only look backwards up to three days. No sense in involving every index in our search. CLIENT_IP_NAME = "client_ip" TIMESTAMP_NAME = "@timestamp" async def find_top_clients( config: plugins.configuration.BlockyConfiguration, aggtype: typing.Literal["bytes", "requests"] = "requests", duration: str = "12h", no_hits: int = 100, filters: typing.List[str] = None, ) -> typing.List[typing.Tuple[str, int]]: """Finds the top clients (IPs) in the database based on the parameters provided. Searches for the top clients by either traffic volume (bytes) or requests.""" assert aggtype in ["bytes", "requests"], "Only by-bytes or by-requests aggregations are supported" if isinstance(filters, str): filters = [filters] elif filters is None: filters = [] q = elasticsearch_dsl.Search(using=config.elasticsearch) q = q.filter("range", **{TIMESTAMP_NAME: {"gte": f"now-{duration}"}}) # Make a list of the past three days' index names: d = datetime.datetime.utcnow() t = [] for i in range(0, MAX_DB_DAYS): index_name = d.strftime(config.index_pattern) has_index = await config.elasticsearch.indices.exists(index=index_name) if has_index: t.append(index_name) d -= datetime.timedelta(days=1) threes = ",".join(t) if not threes: return [] # Add all search filters for entry in filters: if entry: k, o, v = entry.split(" ", 2) # key, operator, value xq = q.query # Default is to add as search param if o.startswith("!"): # exclude as search param? o = o[1:] xq = q.exclude if o == "=": q = xq("match", **{k: v}) elif o == "~=": q = xq("regexp", **{k: v}) elif o == "==": q = xq("term", **{k: v}) else: raise TypeError(f"Unknown operator {o} in search filter: {entry}") if aggtype == "requests": q.aggs.bucket("requests_per_ip", elasticsearch_dsl.A("terms", field=f"{CLIENT_IP_NAME}.keyword", size=no_hits)) elif aggtype == "bytes": q.aggs.bucket( "requests_per_ip", elasticsearch_dsl.A("terms", field=f"{CLIENT_IP_NAME}.keyword", size=no_hits, order={"bytes_sum": "desc"}), ).metric("bytes_sum", "sum", field="bytes") resp = await config.elasticsearch.search(index=threes, body=q.to_dict(), size=0, timeout="30s") top_ips = [] if "aggregations" not in resp: print(f"Could not find aggregated data. Are you sure the index pattern {config.index_pattern} exists?") return [] for entry in resp["aggregations"]["requests_per_ip"]["buckets"]: if "bytes_sum" in entry: top_ips.append( ( entry["key"], int(entry["bytes_sum"]["value"]), ) ) else: top_ips.append( ( entry["key"], int(entry["doc_count"]), ) ) return top_ips class BanRule: def __init__(self, ruledict): self.description = ruledict["description"] self.aggtype = ruledict["aggtype"] self.limit = ruledict["limit"] self.duration = ruledict["duration"] self.filters = [x.strip() for x in ruledict["filters"].split("\n") if x.strip()] async def list_offenders(self, config: plugins.configuration.BlockyConfiguration): """Find top clients by $metric, see if they cross the limit...""" offenders = [] candidates = [] try: candidates = await find_top_clients(config, aggtype=self.aggtype, duration=self.duration, filters=self.filters) except (asyncio.exceptions.TimeoutError, elasticsearch.exceptions.ConnectionTimeout, elasticsearch.exceptions.ConnectionError): print("Offender search timed out, retrying later!") except elasticsearch.exceptions.TransportError: print("Transport error (503?), retrying later") for candidate in candidates: if candidate[1] >= self.limit: offenders.append(candidate) return offenders async def run(config: plugins.configuration.BlockyConfiguration): # Search forever, sleep a little in between while True: # Find expired rules now = int(time.time()) all_items = [item for item in config.sqlite.fetch("lists", limit=0)] for item in all_items: if item['expires'] == -1: continue # never expires if item['expires'] < now: print(f"Expiring {item['type']} rule for {item['ip']}") if item['type'] == 'allow': config.allow_list.remove(item['ip']) elif item['type'] == 'block': config.block_list.remove(item['ip']) # Try adding a temporary whitelist entry to flush on hosts try: config.allow_list.add( ip=item["ip"], timestamp=now, expires=now + 600, # Expire this rule in 10 minutes reason="Temporary allow-listed by BLocky4 to unblock IP due to block expiring", host=plugins.configuration.DEFAULT_HOST_BLOCK, force=False ) except plugins.lists.BlockListException: pass # If it conflicts, it should already be unblocked, so we don't care. else: print("I don't actually know items of type {item['type']}, ignoring...") all_rules = [item for item in config.sqlite.fetch("rules", limit=0)] for rule in all_rules: # print(f"Running rule #{rule['id']}: {rule['description']}...") my_rule = BanRule(rule) off = await my_rule.list_offenders(config) if off: for offender in off: off_ip = offender[0] off_limit = offender[1] off_ip_na = netaddr.IPAddress(off_ip) ignore_ip = False for allowed_ip in config.allow_list: if ( isinstance(allowed_ip.network, netaddr.IPNetwork) and off_ip_na in allowed_ip.network or isinstance(allowed_ip.network, netaddr.IPAddress) and off_ip_na == allowed_ip.network ): # print(f"IP {off_ip} is on the allow list, ignoring...") ignore_ip = True break for blocked_ip in config.block_list: if ( isinstance(blocked_ip.network, netaddr.IPNetwork) and off_ip_na in blocked_ip.network or isinstance(blocked_ip.network, netaddr.IPAddress) and off_ip_na == blocked_ip.network ): # print(f"IP {off_ip} is already blocked, ignoring...") ignore_ip = True break if not ignore_ip: off_reason = f"{rule['description']} ({off_limit} >= {rule['limit']})" print(f"Found new offender, {off_ip}: {off_reason}") now = int(time.time()) expires = now + config.default_expire_seconds config.block_list.add( ip=off_ip, timestamp=now, expires=expires, reason=off_reason, host=plugins.configuration.DEFAULT_HOST_BLOCK, ) await asyncio.sleep(15)