Improper Input Validation

#!/usr/bin/env python3
# verify_cve_2026_41044.py
# Exit codes: 0=PATCHED, 1=VULNERABLE, 2=UNKNOWN, 3=USAGE

import os
import re
import sys
import zipfile
import subprocess
from pathlib import Path

USAGE = "Usage: python3 verify_cve_2026_41044.py <ACTIVEMQ_HOME|jar-file|version-string>"


def norm_version(v):
    if not v:
        return None
    v = v.strip()
    m = re.search(r'(\d+)\.(\d+)\.(\d+)', v)
    if not m:
        return None
    return tuple(int(x) for x in m.groups())


def version_to_str(vt):
    return '.'.join(str(x) for x in vt) if vt else 'unknown'


def compare(a, b):
    return (a > b) - (a < b)


def is_vulnerable(vt):
    if vt is None:
        return None
    if compare(vt, (6, 0, 0)) >= 0 and compare(vt, (6, 2, 5)) < 0:
        return True
    if compare(vt, (5, 19, 6)) < 0:
        return True
    return False


def read_manifest_version(jar_path):
    try:
        with zipfile.ZipFile(jar_path, 'r') as zf:
            for name in zf.namelist():
                if name.upper() == 'META-INF/MANIFEST.MF':
                    data = zf.read(name).decode('utf-8', errors='ignore')
                    for line in data.splitlines():
                        if line.lower().startswith('implementation-version:'):
                            return line.split(':', 1)[1].strip()
    except Exception:
        return None
    return None


def find_version_in_filenames(root):
    patterns = [
        r'activemq-(?:broker|all|client)-(?P<v>\d+\.\d+\.\d+)\.jar$',
        r'apache-activemq-(?P<v>\d+\.\d+\.\d+)',
        r'activemq-(?P<v>\d+\.\d+\.\d+)'
    ]
    candidates = []
    for base, _, files in os.walk(root):
        for f in files:
            for pat in patterns:
                m = re.search(pat, f)
                if m:
                    candidates.append(m.group('v'))
    if candidates:
        # Prefer highest parsed version found in the tree
        parsed = [norm_version(c) for c in candidates]
        parsed = [p for p in parsed if p is not None]
        if parsed:
            parsed.sort()
            return version_to_str(parsed[-1])
    return None


def run_activemq_version(home):
    bin_candidates = [Path(home) / 'bin' / 'activemq', Path(home) / 'bin' / 'activemq.bat']
    for candidate in bin_candidates:
        if candidate.exists():
            try:
                if candidate.suffix.lower() == '.bat':
                    proc = subprocess.run([str(candidate)], capture_output=True, text=True, timeout=15, shell=True)
                else:
                    proc = subprocess.run([str(candidate), '--version'], capture_output=True, text=True, timeout=15)
                out = (proc.stdout or '') + '\n' + (proc.stderr or '')
                m = re.search(r'(\d+\.\d+\.\d+)', out)
                if m:
                    return m.group(1)
            except Exception:
                pass
    return None


def extract_version(target):
    p = Path(target)

    # Raw version string
    raw = norm_version(target)
    if raw:
        return version_to_str(raw), 'user-supplied version string'

    # Jar file
    if p.is_file() and p.suffix.lower() == '.jar':
        mv = read_manifest_version(str(p))
        if mv:
            return mv, f'manifest in {p.name}'
        m = re.search(r'(\d+\.\d+\.\d+)', p.name)
        if m:
            return m.group(1), f'filename {p.name}'
        return None, f'could not extract version from {p}'

    # Directory / install root
    if p.is_dir():
        v = run_activemq_version(str(p))
        if v:
            return v, f'{p}/bin/activemq'

        lib = p / 'lib'
        if lib.is_dir():
            for jar_name in ['activemq-broker', 'activemq-all', 'activemq-client']:
                jars = sorted(lib.glob(f'{jar_name}-*.jar'))
                for jar in jars:
                    mv = read_manifest_version(str(jar))
                    if mv:
                        return mv, f'manifest in {jar.name}'
                    m = re.search(r'(\d+\.\d+\.\d+)', jar.name)
                    if m:
                        return m.group(1), f'filename {jar.name}'

        fv = find_version_in_filenames(str(p))
        if fv:
            return fv, f'filenames under {p}'

        return None, f'could not determine version from {p}'

    return None, 'target path not found'


def main():
    if len(sys.argv) != 2:
        print('UNKNOWN - ' + USAGE)
        sys.exit(3)

    target = sys.argv[1]
    version_str, source = extract_version(target)
    vt = norm_version(version_str) if version_str else None

    if vt is None:
        print(f'UNKNOWN - Unable to determine ActiveMQ version ({source})')
        sys.exit(2)

    vuln = is_vulnerable(vt)
    if vuln is None:
        print(f'UNKNOWN - Parsed version {version_to_str(vt)} but could not classify')
        sys.exit(2)

    if vuln:
        print(f'VULNERABLE - ActiveMQ version {version_to_str(vt)} from {source} matches CVE-2026-41044 affected ranges (<5.19.6 or 6.0.0-6.2.4)')
        sys.exit(1)
    else:
        print(f'PATCHED - ActiveMQ version {version_to_str(vt)} from {source} is outside the CVE-2026-41044 affected ranges')
        sys.exit(0)


if __name__ == '__main__':
    main()