Filters the IP addresses from HTTP verb (for nginx/apache) and checks if the IP address is within the allowed IP address list Args: pattern_type: A string to select the correct rule and ip found: A matching regex string instant_ban: Boole
(self, pattern_type, found, instant_ban)
| 64 | time.sleep(0.0001) # Ensure it doesnt kill CPU |
| 65 | |
| 66 | def filter(self, pattern_type, found, instant_ban): |
| 67 | """ |
| 68 | Filters the IP addresses from HTTP verb (for nginx/apache) and |
| 69 | checks if the IP address is within the allowed IP address list |
| 70 | |
| 71 | Args: |
| 72 | pattern_type: A string to select the correct rule and ip |
| 73 | found: A matching regex string |
| 74 | instant_ban: Boolean passed to instantly ban the IP on a certain regex match |
| 75 | """ |
| 76 | |
| 77 | cond = pattern_type in ("apache", "nginx") |
| 78 | |
| 79 | ip = found[not cond] |
| 80 | t = datetime.strptime(found[cond], self.rules[pattern_type]["time_format"]) |
| 81 | |
| 82 | if cond and int(found[3]) not in self.rules[pattern_type]["http_status_blocks"]: |
| 83 | return |
| 84 | |
| 85 | this_year = datetime.now().year |
| 86 | |
| 87 | if t.year != this_year: |
| 88 | t = t.replace(year=this_year) # Assume the request was this year |
| 89 | |
| 90 | ip_type = self.__check_ip(ip) |
| 91 | |
| 92 | if not ip_type: |
| 93 | ip = socket.gethostbyname(ip) |
| 94 | ip_type = self.__check_ip(ip) |
| 95 | |
| 96 | if ip not in self.settings["ignored_ips"]: |
| 97 | if instant_ban: |
| 98 | if self.database_connection.select(ip) is not None: |
| 99 | return |
| 100 | |
| 101 | log_msg = "IP: {} has been blacklisted and the firewall rules have been updated." \ |
| 102 | " Acquired an instant ban via {}.\n".format(ip, pattern_type) |
| 103 | |
| 104 | if self.log_settings["active"]: |
| 105 | self.log(log_msg) |
| 106 | print(log_msg, end='') |
| 107 | |
| 108 | return self.blacklist(ip, log_msg=log_msg, ip_type=ip_type) |
| 109 | |
| 110 | if ip not in self.ip_dict[pattern_type]: |
| 111 | self.ip_dict[pattern_type][ip] = {"amount": 0, "last_request": None} |
| 112 | self.check(ip, pattern_type, t, ip_type) |
| 113 | |
| 114 | def check(self, ip, pattern_type, time_object, ip_type="v4"): |
| 115 | """ |
no test coverage detected