^^&&^^

Edit File: i360-storage-new

#!/opt/imunify360/venv/bin/python
#
# i360-storage-new      Python script to compile rules for appropriate
#                       proactive module version.
#

import argparse
import os
from os import path
import logging
import time
import shutil
import subprocess
import sys
import tempfile
import sentry_sdk
import json

VERSION = '8.5'
I360_PHP_OPTS="/usr/share/i360-php-opts/"
SRC = "/var/imunify360/files/proactive/rules/%s" % VERSION
DEST = path.join(I360_PHP_OPTS, "sigs", VERSION)
RULES_FILE = path.join(I360_PHP_OPTS, "rules.yaml")
TOOL = "/usr/bin/i360-storage.v2"
MUST_HAVE_ITEMS = ('params_pattern_list',
                   'rce_patterns_list')
RESULT_V1 = [ '.rules', '.rulesdump' ]
RESULT_V2 = '.rules.v2'
TOOL_OLD = "/usr/bin/i360-storage.immunity"
# send exceptions to php-daemon's project
SENTRY_DSN = "https://426c8403a5084f1081917cd63932fe7f@im360.sentry.cloudlinux.com/8"
IMUNIFY360_TAGS = "/var/imunify360/.sentry_tags"

def must_have_items_check(silently_skip):
    if not path.exists(SRC):
        sys.exit(0 if silently_skip else
                 "ERROR: directory '%s' does not EXIST" % SRC)

for item in MUST_HAVE_ITEMS:
        if not path.exists(path.join(SRC, item)):
            sys.exit(0 if silently_skip else
                     "ERROR: file '%s' is NOT FOUND in '%s' directory" % (item, SRC))

def sentry_init():
    sentry_sdk.init(dsn=SENTRY_DSN)
    with sentry_sdk.configure_scope() as scope:
        scope.set_tag("script_name", "i360-storage-new")
        scope.set_tag("proactive_version", VERSION)

# load sentry tags from file, pass on Exception
        try:
            with open(IMUNIFY360_TAGS, "r") as f:
                tags = json.loads(f.read())
                skip = {"name", "firewall", "strategy"}
                for k, v in tags.items():
                    if not k or k in skip:
                        continue
                    key = "imunify360_version" if k == "version" else k
                    scope.set_tag(key, v)
        except Exception as e:
            pass

def flush_sentry():
    client = sentry_sdk.Hub.current.client
    if client is not None:
        client.flush(timeout=1.0)

if __name__ == '__main__':
    try:
        sentry_init()
    except Exception:
        pass

ts_begin = time.time()
    logging.basicConfig(filename='/var/log/proactive_defense_compiler.log',
      filemode='w', level=logging.DEBUG, format='%(asctime)s %(levelname)-8s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
    parser = argparse.ArgumentParser(description="compile PA rules [%s -> %s]" % (SRC, DEST))
    parser.add_argument("--silent-on-noop",
                        action="store_true",
                        help="do nothing if %s does not exist" % SRC)
    parser.add_argument("--optimize",
                        action="store_true",
                        default=False,
                        help="skip v1 compilation if v2 is available")
    args = parser.parse_args()

skip_v1 = args.optimize
    must_have_items_check(args.silent_on_noop)
    tempdir = tempfile.mktemp(prefix='%s_tmpdir_' % path.basename(DEST),
                              dir=path.dirname(DEST))
    shutil.copytree(SRC, tempdir)
    try:
        # Add (overwrite) rules.yaml from /usr/share/i360-php-opts
        shutil.copy2(RULES_FILE, tempdir)
    except Exception as e:
        sentry_sdk.capture_exception(e)
        flush_sentry()
        sys.exit("ERROR: %s" % e)

os.chdir(tempdir)
    os.umask(0o022)

result = []
    try:
        output = subprocess.check_output([TOOL, tempdir + '/.rules.v2', tempdir], stderr=subprocess.STDOUT, universal_newlines=True)
    except subprocess.CalledProcessError as e:
        # full backtrace is not needed for tool call error
        logging.critical('%s\n==================================' % e.output)
        if e.returncode == 2:
            skip_v1 = False
            print("ERROR: %s" % e)
        else:
            sys.exit("ERROR: %s" % e)
    else:
        result.append(RESULT_V2)

if skip_v1 == False:
        try:
            output = subprocess.check_output(
                        [TOOL_OLD, 'mkbin'],
                        stderr=subprocess.STDOUT,
                        universal_newlines=True
                    )
            output = subprocess.check_output(
                        [TOOL_OLD,
                         'mkdump',
                         '.rules',
                         '.rulesdump'],
                        stderr=subprocess.STDOUT,
                        universal_newlines=True
                    )
        except subprocess.CalledProcessError as e:
            # full backtrace is not needed for tool call error
            logging.critical(
                    '%s\n==================================' % e.output
                    )
            sys.exit("ERROR: %s" % e)
        result.extend(RESULT_V1)

os.makedirs(DEST, mode=0o755, exist_ok=True)
    for item in result:
        # get ready for atomic rewrite (copy + rename)
        tmpfn = tempfile.mktemp(suffix='_i360tmp', prefix=item, dir=DEST)
        # copy data and mode bits
        if os.path.exists(path.join(tempdir, item)):
            shutil.copy(path.join(tempdir, item), tmpfn)

# that is Ok for .rulesdump + .rules to be handled not as a single step,
            # as far so far .rules is a fallback when we fail to mmap() .rulesdump
            os.rename(tmpfn, path.join(DEST, item))

os.chdir(tempfile.gettempdir())  # chdir away from tempdir
    shutil.rmtree(tempdir)
    ts_end = time.time()
    logging.info("Success. Execution time %d s" % (ts_end - ts_begin))
    print("Success.")