Have you Patched the Apache Log4j vulnerability CVE-2021-44228?

#!/usr/bin/env python3

import os
import sys
import copy
import gzip
import urllib.parse
import argparse
from datetime import datetime, timedelta
from collections import defaultdict
import traceback
import base64
import time
import subprocess

DEFAULT_PATHS = [‘/var/log’, ‘/storage/log/vmware’, ‘/var/atlassian/application-data/jira/log’, ‘/home/logs’]
LINUX_PATH_SKIPS_START = set([“/proc”, “/dev”, “/sys/kernel/debug”, “/sys/kernel/slab”, “/sys/devices”, “/usr/src/linux”])

class log4j2_detect(object):

    EXPLOIT_STRINGS = [‘${jndi:ldap:’, ‘${jndi:rmi:/’, ‘${jndi:ldaps:/’, ‘${jndi:dns:/’, ‘${jndi:nis:/’, ‘${jndi:nds:/’, ‘${jndi:corba:/’, ‘${jndi:iiop:/’, ‘$%7Bjndi:ldap://’, ‘%2524%257Bjndi:ldap:/’, ‘%2F%252524%25257Bjndi%3Aldap%3A%2F’, ‘%2F%252524%25257Bjndi%3Aldaps%3A%2F’, ‘%2F%252524%25257Bjndi%3Adns%3A%2F’
    ‘%2F%252524%25257Bjndi%3Armi%3A%2F’, ‘${jndi:ldap:’, ‘${jndi:rmi:’, ‘${jndi:ldaps:’, ‘${jndi:dns:’, ‘${jndi:nis:’, ‘${jndi:nds:’, ‘${jndi:corba:’, ‘${jndi:iiop:’, ‘${ctx:’]
    INSTALL_PATHS = [‘/usr/local’]
    PLAIN_STRINGS = {
        “https://gist.github.com/Neo23x0/e4c8b03ff8cdf1fa63b7d15db6e3860b#gistcomment-3991502”: [
            ” header with value of BadAttributeValueException: “
        ],
        “https://gist.github.com/Neo23x0/e4c8b03ff8cdf1fa63b7d15db6e3860b#gistcomment-3991700”: [
            “at java.naming/com.sun.jndi.url.ldap.ldapURLContext.lookup(“,
            “.log4j.core.lookup.JndiLookup.lookup(JndiLookup”
        ],
        “https://github.com/Neo23x0/log4shell-detector/issues/5#issuecomment-991963675”: [
            ‘${base64:JHtqbmRp’
        ],
        “https://github.com/tangxiaofeng7/CVE-2021-44228-Apache-Log4j-Rce/issues/1”: [
            ‘Reference Class Name: foo’
        ]
    }

    def __init__(self, maximum_distance, debug, fast):
        self.prepare_detections(maximum_distance)
        self.debug = debug
        self.fast = fast
        self.number_of_detections = 0
        self.affected_files = []

    def decode_line(self, line):
        while “%” in line:
            line_before = line
            line = urllib.parse.unquote(line)
            if line == line_before:
                break
        return line

    def evaluate_log_paths(self):
        paths = []
        print(“Automatically evaluating the folders to which Applications write logsn”)
        command = “lsof 2>/dev/null | grep ‘\.log’ | sed ‘s/.* \//\//g’ | sort | uniq”
        path_eval = subprocess.Popen(command,shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
        output = path_eval.communicate()[0].splitlines()
        for o in output:
            path = os.path.dirname(o)
            if isinstance(path, bytes):
                path = path.decode(“utf-8”)

            # Some filters
            skip_append = False
            # If already in list – skip
            if path in paths:
                skip_append = True
            # If in exclude list – skip
            for exclude in LINUX_PATH_SKIPS_START:
                if path.startswith(exclude):
                    skip_append = True
            if skip_append:
                continue

            # Append the found path
            paths.append(path)
            print(“Adding PATH: %s” % path)

        return paths

    def log4j_detect(self):
        “””
        Detect the version of log4j instance
        in the default install paths
        “””
        checker_commands = [
        “ps aux | egrep ‘[l]og4j'”,
        “find / -iname “log4j*””,
        “lsof | grep log4j”,
        “grep -r –include *.[wj]ar “JndiLookup.class” / 2>&1 | grep matches”,
        ]
        for checker_command in checker_commands:
            if len(subprocess.Popen(checker_command,shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT).communicate()[0].splitlines()) > 0:
                return True
        return False

    def base64_decode(self, log):
        payload = “”
        if “Base64” in log:
            payload = base64.decodebytes(log.split(“Base64/”)[1].split(“}”)[0].encode()).decode()
        return payload

    def check_line(self, line):
        “””
        Check line and identify presence of
        a character present in exploit string.
        Look for the next consecutive character
        within the maximum distance
        :param line:    single line from a log file
        :type  line:    str
        :param detection_pad:    dictionary created from attack string
        :type  detection_pad:    dict
        :return:    detection string if found
        :rtype:     str
        “””

        decoded_line = self.decode_line(line)

        for ref, strings in self.PLAIN_STRINGS.items():
            for s in strings:
                if s in line or s in decoded_line:
                    return s

        decoded_line = decoded_line.lower()
        linechars = list(decoded_line)
        # temporary detection pad
        dp = copy.deepcopy(self.detection_pad)
        # Walk over characters
        for c in linechars:
            for exploit_string in dp:
                # If the character in the line matches the character in the detection
                if c == dp[exploit_string][“chars”][dp[exploit_string][“level”]]:
                    dp[exploit_string][“level”] += 1
                    dp[exploit_string][“current_distance”] = 0
                # If level > 0 count distance to the last char
                if dp[exploit_string][“level”] > 0:
                    dp[exploit_string][“current_distance”] += 1
                    # If distance is too big, reset level to zero
                    if dp[exploit_string][“current_distance”] > dp[exploit_string][“maximum_distance”]:
                       dp[exploit_string][“current_distance”] = 0
                       dp[exploit_string][“level”] = 0
                # Is the pad completely empty?
                if len(dp[exploit_string][“chars”]) == dp[exploit_string][“level”]:
                    return exploit_string

    def scan_file(self, file_path):
        “””
        Give a path, scan all files to check if
        an exploit string is written/obfuscated
        in it via detection pads
        :param file_path:    single file in the path
        :type  file_path:    str
        :return:    matches in file
        :rtype:     dict
        “””
        matches_in_file = []
        try:
            # Gzipped logs
            if “log.” in file_path and file_path.endswith(“.gz”):
                with gzip.open(file_path, ‘rt’) as gzlog:
                    c = 0
                    for line in gzlog:
                        c += 1
                        # Quick mode – timestamp check
                        if self.fast and not “2021” in line and not “2022” in line:
                            continue
                        # Analyze the line
                        result = self.check_line(line)
                        if “${ctx:” in str(result):
                            print(“Possible DOS/RCE exploitation for new CVEsn”)
                        if result:
                            payload = self.base64_decode(line.rstrip())
                            self.number_of_detections+=1
                            matches_dict = {
                                “line_number”: c,
                                “match_string”: result,
                                “line”: line.rstrip()
                            }
                            print(“nWarning: Exploitation attempt detected FILE: %s LINE_NUMBER: %d LINE: %s DEOBFUSCATED_STRING: %s DECODED_PAYLOAD: %s” %
                            (file_path, matches_dict[‘line_number’], matches_dict[‘line’], result, payload))
                            matches_in_file.append(matches_dict)
            # Plain Text
            else:
                with open(file_path, ‘r’) as logfile:
                    c = 0
                    for line in logfile:
                        c += 1
                        # Quick mode – timestamp check
                        if self.fast and not “2021” in line and not “2022” in line:
                            continue
                        # Analyze the line
                        result = self.check_line(line)
                        if “${ctx:” in str(result):
                            print(“Possible DOS/RCE exploitation for new CVEsn”)
                        if result:
                            payload = self.base64_decode(line.rstrip())
                            self.number_of_detections+=1
                            matches_dict = {
                                “line_number”: c,
                                “match_string”: result,
                                “line”: line.rstrip()
                            }
                            print(“nWarning: Exploitation attempt detected FILE: %s LINE_NUMBER: %d LINE: %s DEOBFUSCATED_STRING: %s DECODED_PAYLOAD: %s” %
                            (file_path, matches_dict[‘line_number’], matches_dict[‘line’], result, payload))
                            matches_in_file.append(matches_dict)

        except UnicodeDecodeError as e:
            if self.debug:
                print(“nUnable to process FILE: %s REASON: most likely not an ASCII based log file” % file_path)
        except Exception as e:
            print(“nUnable to process FILE: %s REASON: %s” % (file_path, traceback.print_exc()))

        return matches_in_file

    def scan_path(self, path):
        “””
        Give a path, scan all files to check if
        an exploit string is written/obfuscated
        in it via detection pads
        :param path:    single line from a log file
        :type  path:    str
        :return:    number of detections
        :rtype:     int
        “””
        affected_files = self.affected_files
        detections = defaultdict(lambda: defaultdict())
        # Loop over files
        for root, directories, files in os.walk(path, followlinks=False):
            for filename in files:
                file_path = os.path.join(root, filename)
                print(“Processing file : %s” % file_path)
                if self.debug:
                    print(“Processing file :  %s” % file_path)

                evidences_in_file = self.scan_file(file_path)

                if(len(evidences_in_file))>0:
                    affected_files.append(file_path)
                    for evidence in evidences_in_file:
                        detections[file_path][evidence[‘line_number’]] = [evidence[‘line’], evidence[‘match_string’]]

        count_affected_files = len(detections.keys())
        number_of_detections = self.number_of_detections

        if number_of_detections > 0:
            print(“nWarning: %d exploitation attempts detected in PATH: %s” % (number_of_detections, path))
        else:
            print(“nNo Log4Shell exploitation attempts detected in path PATH: %s” % path)

        self.write_out_summary(path, affected_files)

        return number_of_detections

    def prepare_detections(self, maximum_distance):
        “””
        Deconstruct exploit string
        to form detection pads
        :param dist:    maximum distance
        :type  dist:    int
        :return:    detection pads
        :rtype:     dict
        “””
        self.detection_pad = {}
        for ex in self.EXPLOIT_STRINGS:
            self.detection_pad[ex] = {}
            self.detection_pad[ex] = {
                “chars”: list(ex),
                “maximum_distance”: maximum_distance,
                “current_distance”: 0,
                “level”: 0
            }

    def write_out_summary(self, path, affected_files):
        filename = “infected_files_”+str(time.time())+”.txt”
        with open(filename, ‘w’) as f:
            f.writelines(“Affected Files in path : “+path)
            f.write(‘n’)
            f.write(‘**************************************************************’)
            for file in affected_files:
                f.writelines(file)
                f.write(“n”)

if __name__ == ‘__main__’:

    parser = argparse.ArgumentParser(description=’Log4Shell Exploitation Detection’)
    parser.add_argument(‘-p’, nargs=’+’, help=’Path to scan’, metavar=’path’, default=”)
    parser.add_argument(‘-d’, help=’Maximum distance between each character, Default=30′, metavar=’distance’, default=30)
    parser.add_argument(‘–fast’, action=’store_true’, default=False, help=”Skip log lines that don’t contain a 2021 or 2022 time stamp”)
    parser.add_argument(‘–defaultpaths’, action=’store_true’, default=False, help=’Scan a set of default paths known to contain relevant log files.’)
    parser.add_argument(‘–debug’, action=’store_true’, default=False, help=’Debug output’)

    args = parser.parse_args()

    detective = log4j2_detect(maximum_distance=args.d, debug=args.debug, fast=args.fast)

    if not args.p and not args.defaultpaths:
        parser.print_help(sys.stderr)
        print(“”)
        print(“Must specify at least one folder to scan with -p target-folder or use –defaultpaths”)
        sys.exit(1)

    print(“”)
    date_scan_start = datetime.now()
    print(“Scan initiated at DATE: %s” % date_scan_start)
    check = detective.log4j_detect()

    # Prepare the detection pads
    #detection_pad = prepare_detections(int(args.d))

    # Counter
    all_detections = 0

    # Scan paths
    paths = args.p
    if args.defaultpaths:
        paths = detective.evaluate_log_paths()
    for path in paths:
        if not os.path.isdir(path):
            print(“Path %s doesn’t exist” % path)
            if not args.defaultpaths:
                print(“Path %s doesn’t exist” % path)
            continue
        print(“Scanning FOLDER: %s …” % path)
        detections = detective.scan_path(path)
        all_detections += detections

    # Finish
    if all_detections > 0:
        print(“nWarning: %d exploitation attempts detected in the complete scan” % all_detections)
    else:
        print(“nNo exploitation attempts detected in the scan”)
    date_scan_end = datetime.now()
    print(“Scan completed DATE: %s” % date_scan_end)