Mr.X

import asyncio
import datetime as DT
import logging
import time
from ipaddress import ip_network

import peewee

from defence360agent import utils
from defence360agent.contracts import messages, plugins
from defence360agent.model import simplification
from im360.api import ips
from im360.contracts import config
from im360.model import firewall
from im360.simple_rpc.resident_socket import send_to_socket

logger = logging.getLogger(__name__)
IP_LIST_UPDATE_TIMEOUT = 60  # seconds


class WhitelistCurrentUser(plugins.MessageSink):
    async def create_sink(self, loop):
        self._loop = loop

    @plugins.expect(messages.MessageType.CommandInvoke)
    async def whitelist_current_user_ip(self, message):
        try:
            ip = ip_network(message["params"].get("remote_addr"))
        except ValueError:
            return  # do nothing for invalid IP

        if ip.version == 6:
            ip = ip_network((int(ip.network_address), 64), strict=False)
        is_root = not message["params"].get("user")
        await add_to_whitelist(ip, full_access=is_root, loop=self._loop)


@utils.timed_cache(DT.timedelta(minutes=15))
async def add_to_whitelist(ip, full_access, *, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()
    logger.info("Adding %s to whitelist with ttl", ip)
    try:
        ip_obj = await simplification.run_in_executor(
            loop, lambda: firewall.IPList.get(ip=ip)
        )
    except peewee.DoesNotExist:
        pass
    else:
        if (
            ip_obj.listname == firewall.IPList.WHITE
        ) and not ip_obj.auto_whitelisted:
            # do not replace manually whitelisted
            return
        else:
            # Either expired BLACK/GRAY/GRAY_SPLASHSCREEN lists record
            # or auto-whitelisted, replacing
            await ips.IPApi.unblock((ip_obj.ip_network,), ip_obj.listname)

    new_expiration = (
        0
        if config.AutoWhiteList.ttl_value() == 0
        else int(time.time() + config.AutoWhiteList.ttl_value())
    )

    comment = "IP auto-whitelisted with "
    if new_expiration == 0:
        comment += "no expiration"
    else:
        comment += "expiration date: {:%Y-%m-%d %H:%M:%S}".format(
            DT.datetime.fromtimestamp(new_expiration)
        )
    await send_to_socket(
        msg={
            "method": "IP_LISTS_UPDATE",
            "action": "add",
            "purpose": firewall.Purpose.WHITE,
            "items": {
                "items": [str(ip)],
                "expiration": new_expiration,
                "full_access": full_access,
                "comment": comment,
                "auto_whitelisted": True,
            },
        },
        timeout=IP_LIST_UPDATE_TIMEOUT,
    )
    logger.info("Added %s to whitelist with expiration %s", ip, new_expiration)

Name	Type	Size	Permission
__pycache__	Folder		0755
__init__.py	File	0 B	0644
coraza_ff_watcher.py	File	2.82 KB	0644
cpanel_uploader.py	File	2.71 KB	0644
export_wblist.py	File	4.14 KB	0644
fgw.py	File	1.71 KB	0644
lfd.py	File	3.43 KB	0644
modsec_ruleset_checker.py	File	4.91 KB	0644
ossec_rules_checker.py	File	1.76 KB	0644
pam_manager.py	File	6.59 KB	0644
php_immunity.py	File	1.4 KB	0644
remoteip_install.py	File	1.03 KB	0644
repeater.py	File	1.16 KB	0644
send_server_config.py	File	4.43 KB	0644
service_manager.py	File	5.22 KB	0644
startup_actions.py	File	2.42 KB	0644
strategy_getter.py	File	2 KB	0644
waf_rules_configurator.py	File	2.48 KB	0644
webshield_feature_flags.py	File	1.12 KB	0644
whitelist_current_user.py	File	2.75 KB	0644

Upload:

Command:

Filemanager

Server Info

System Info