YANZ UPLOAD ANJG

#!/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 import yaml VERSION = '8.8' I360_PHP_OPTS="/usr/share/i360-php-opts/" SRC = "/var/imunify360/files/proactive/rules/%s" % VERSION DEST = path.join(I360_PHP_OPTS, "sigs", VERSION) # both compilers look for exactly this name in the work directory RULES_YAML = "rules.yaml" 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)) class RulesYamlError(Exception): """rules.yaml is absent, unparseable or fails the schema checks.""" class RulesCompileError(Exception): """The rule sources in the work directory could not be compiled.""" def load_rules_yaml(rules_file): try: with open(rules_file) as fh: doc = yaml.safe_load(fh) except (OSError, IOError, UnicodeDecodeError, yaml.YAMLError) as e: raise RulesYamlError("cannot read '%s': %s" % (rules_file, e)) if not isinstance(doc, dict): raise RulesYamlError("'%s' does not contain a YAML mapping" % rules_file) return doc def rules_yaml_signs(doc, rules_file): # `signs` is the YYMMDD date of the rules release. Both compilers read it as # an integer (C parse_long, Go strconv.Atoi), so newness is an int compare. if 'signs' not in doc: raise RulesYamlError("'%s' has no 'signs' value" % rules_file) try: return int(str(doc['signs']).strip()) except ValueError: raise RulesYamlError("'%s' has a non-integer 'signs' value %r" % (rules_file, doc['signs'])) def validate_rules_yaml(doc, rules_file): """Reject candidates the compilers accept but that carry no detection.""" rules = doc.get('rules') if not rules: raise RulesYamlError("'%s' has no 'rules' section" % rules_file) if not isinstance(rules, list): raise RulesYamlError("'%s' has a malformed 'rules' section" % rules_file) def prefer_cdn_rules(cdn_rules, bundled_rules): """Decide whether the CDN-delivered rules.yaml should be compiled instead of the RPM-bundled copy. Returns (use_cdn, reason); reason explains a refusal. """ try: bundled_signs = rules_yaml_signs(load_rules_yaml(bundled_rules), bundled_rules) except RulesYamlError as e: # without a comparable bundled signs we cannot prove the CDN copy newer return False, 'keeping bundled rules.yaml: %s' % e try: doc = load_rules_yaml(cdn_rules) cdn_signs = rules_yaml_signs(doc, cdn_rules) if cdn_signs <= bundled_signs: raise RulesYamlError('signs %d is not newer than bundled signs %d' % (cdn_signs, bundled_signs)) validate_rules_yaml(doc, cdn_rules) except RulesYamlError as e: return False, 'using bundled rules.yaml: %s' % e return True, None def install_bundled_rules(bundled_rules, tempdir): try: shutil.copy2(bundled_rules, path.join(tempdir, RULES_YAML)) except Exception as e: sentry_sdk.capture_exception(e) flush_sentry() sys.exit("ERROR: %s" % e) def compile_rules(tempdir, skip_v1): """Run the v2 (and, unless skipped, the v1) compiler over tempdir.""" result = [] try: subprocess.check_output([TOOL, path.join(tempdir, RESULT_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: # no usable Hyperscan here: v1 only. Not a defect in the rules. skip_v1 = False print("ERROR: %s" % e) else: raise RulesCompileError(e) else: result.append(RESULT_V2) if skip_v1 == False: try: subprocess.check_output( [TOOL_OLD, 'mkbin'], stderr=subprocess.STDOUT, universal_newlines=True ) 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 ) raise RulesCompileError(e) result.extend(RESULT_V1) return result def compile_preferring_cdn(tempdir, bundled_rules, skip_v1): """Compile the rules in tempdir, preferring the CDN-delivered rules.yaml already copied there over the RPM-bundled copy, but only when it is newer, schema-valid and actually compiles. Falls back to the bundled copy. """ cdn_rules = path.join(tempdir, RULES_YAML) cdn_present = path.exists(cdn_rules) use_cdn, reason = prefer_cdn_rules(cdn_rules, bundled_rules) if not use_cdn: # an absent CDN copy is the normal state, not something to warn about log = logging.warning if cdn_present else logging.debug log('rules.yaml: %s', reason) install_bundled_rules(bundled_rules, tempdir) return compile_rules(tempdir, skip_v1) logging.info('rules.yaml: compiling the CDN copy from %s', tempdir) try: return compile_rules(tempdir, skip_v1) except RulesCompileError as e: logging.warning('rules.yaml: using bundled rules.yaml: the CDN copy ' 'failed to compile: %s', e) install_bundled_rules(bundled_rules, tempdir) return compile_rules(tempdir, skip_v1) 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) os.chdir(tempdir) os.umask(0o022) try: result = compile_preferring_cdn(tempdir, RULES_FILE, skip_v1) except RulesCompileError as e: sys.exit("ERROR: %s" % e) 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.")