Apache Log4j2 2

#!/usr/bin/env python3
# log4j_check.py
# Scan a directory tree for log4j-core versions inside JAR/WAR/EAR/ZIP archives.
# This script uses the *practical safe target* logic for Log4Shell-era risk:
#   PATCHED = 2.17.0+ (Java 8+), 2.12.3+ on the 2.12.x line, or 2.3.1+ on the 2.3.x line
#   VULNERABLE = versions in the Log4Shell / incomplete-fix window, including 2.15.0 and 2.12.2
# Output: VULNERABLE / PATCHED / UNKNOWN
# Exit codes: 0=PATCHED, 1=VULNERABLE, 2=UNKNOWN

import io
import os
import re
import sys
import zipfile
from typing import List, Tuple, Optional

ARCHIVE_EXTS = {'.jar', '.war', '.ear', '.zip'}
POM_PATH = 'META-INF/maven/org.apache.logging.log4j/log4j-core/pom.properties'
MANIFEST_PATH = 'META-INF/MANIFEST.MF'
VERSION_RE = re.compile(r'^version\s*=\s*(.+)$', re.MULTILINE)
IMPL_VERSION_RE = re.compile(r'^Implementation-Version:\s*(.+)$', re.MULTILINE)

class Version:
    def __init__(self, raw: str):
        self.raw = raw.strip()
        self.main, self.stage_rank, self.stage_num = self._parse(self.raw)

    def _parse(self, s: str):
        s = s.strip().lower()
        m = re.match(r'^(\d+)\.(\d+)(?:\.(\d+))?(?:[.-]?(alpha|beta|rc)(\d+)?)?$', s)
        if not m:
            nums = [int(x) for x in re.findall(r'\d+', s)]
            while len(nums) < 3:
                nums.append(0)
            return tuple(nums[:3]), 3, 0
        major = int(m.group(1))
        minor = int(m.group(2))
        patch = int(m.group(3) or 0)
        stage = m.group(4)
        stage_num = int(m.group(5) or 0)
        rank_map = {'alpha': 0, 'beta': 1, 'rc': 2, None: 3}
        return (major, minor, patch), rank_map[stage], stage_num

    def _cmp_tuple(self):
        return self.main + (self.stage_rank, self.stage_num)

    def __lt__(self, other):
        return self._cmp_tuple() < other._cmp_tuple()

    def __le__(self, other):
        return self._cmp_tuple() <= other._cmp_tuple()

    def __ge__(self, other):
        return self._cmp_tuple() >= other._cmp_tuple()

    def __gt__(self, other):
        return self._cmp_tuple() > other._cmp_tuple()

    def __eq__(self, other):
        return self._cmp_tuple() == other._cmp_tuple()


def in_range(v: Version, low: str, high: str) -> bool:
    return Version(low) <= v < Version(high)


def is_practically_safe(vs: str) -> bool:
    v = Version(vs)
    # Safe targets for real-world Log4j remediation, not just the first 44228 fix.
    if v.main[0] != 2:
        return False
    if v.main[1] == 3:
        return v >= Version('2.3.1')
    if v.main[1] == 12:
        return v >= Version('2.12.3')
    return v >= Version('2.17.0')


def is_vulnerable(vs: str) -> bool:
    v = Version(vs)
    # Practical vulnerable set for Log4Shell-era risk:
    # 2.0-beta9 <= v < 2.3.1
    # 2.4 <= v < 2.12.3
    # 2.13.0 <= v < 2.17.0
    return (
        in_range(v, '2.0-beta9', '2.3.1') or
        in_range(v, '2.4.0', '2.12.3') or
        in_range(v, '2.13.0', '2.17.0')
    )


def read_version_from_zip(zf: zipfile.ZipFile) -> Optional[str]:
    # Preferred source: pom.properties
    try:
        with zf.open(POM_PATH) as f:
            text = f.read().decode('utf-8', errors='ignore')
            m = VERSION_RE.search(text)
            if m:
                return m.group(1).strip()
    except KeyError:
        pass
    except Exception:
        pass

    # Fallback: manifest
    try:
        with zf.open(MANIFEST_PATH) as f:
            text = f.read().decode('utf-8', errors='ignore')
            if 'log4j' in text.lower():
                m = IMPL_VERSION_RE.search(text)
                if m:
                    return m.group(1).strip()
    except KeyError:
        pass
    except Exception:
        pass
    return None


def scan_zip_bytes(data: bytes, logical_name: str, findings: List[Tuple[str, str]], errors: List[str]):
    try:
        with zipfile.ZipFile(io.BytesIO(data)) as zf:
            ver = read_version_from_zip(zf)
            if ver:
                findings.append((logical_name, ver))
            for name in zf.namelist():
                lower = name.lower()
                _, ext = os.path.splitext(lower)
                if ext in ARCHIVE_EXTS:
                    try:
                        nested = zf.read(name)
                        scan_zip_bytes(nested, logical_name + '!' + name, findings, errors)
                    except Exception as e:
                        errors.append(f'nested read failed: {logical_name}!{name}: {e}')
    except Exception as e:
        errors.append(f'zip open failed: {logical_name}: {e}')


def scan_path(root: str):
    findings: List[Tuple[str, str]] = []
    errors: List[str] = []
    for base, _, files in os.walk(root):
        for fn in files:
            full = os.path.join(base, fn)
            _, ext = os.path.splitext(fn.lower())
            if ext not in ARCHIVE_EXTS:
                continue
            try:
                with open(full, 'rb') as f:
                    scan_zip_bytes(f.read(), full, findings, errors)
            except Exception as e:
                errors.append(f'file read failed: {full}: {e}')
    return findings, errors


def main():
    if len(sys.argv) != 2:
        print('UNKNOWN - usage: python3 log4j_check.py <path>')
        sys.exit(2)

    root = sys.argv[1]
    if not os.path.exists(root):
        print(f'UNKNOWN - path does not exist: {root}')
        sys.exit(2)

    findings, errors = scan_path(root)

    if not findings and errors:
        print('UNKNOWN - scan errors prevented reliable assessment')
        for e in errors[:20]:
            print(e)
        sys.exit(2)

    if not findings:
        print('UNKNOWN - no log4j-core artifacts found')
        sys.exit(2)

    vulnerable = []
    patched = []
    unknown_versions = []

    for path, ver in findings:
        try:
            if is_vulnerable(ver):
                vulnerable.append((path, ver))
            elif is_practically_safe(ver):
                patched.append((path, ver))
            else:
                unknown_versions.append((path, ver))
        except Exception:
            unknown_versions.append((path, ver))

    if vulnerable:
        print('VULNERABLE')
        for path, ver in vulnerable[:100]:
            print(f'{path} :: {ver}')
        if patched:
            print(f'INFO - also found {len(patched)} patched artifact(s)')
        if unknown_versions:
            print(f'INFO - also found {len(unknown_versions)} artifact(s) with unclassified version strings')
        sys.exit(1)

    if patched and not unknown_versions:
        print('PATCHED')
        for path, ver in patched[:100]:
            print(f'{path} :: {ver}')
        sys.exit(0)

    print('UNKNOWN - only unclassified versions found or mixed non-vulnerable artifacts without clear safe status')
    for path, ver in (unknown_versions or patched)[:100]:
        print(f'{path} :: {ver}')
    sys.exit(2)


if __name__ == '__main__':
    main()