Apache Fory PyFory ReduceSerializer deserialization policy bypass

#!/usr/bin/env python3
# CVE-2026-48207 verifier for Apache Fory PyFory
# Exit codes: 0=PATCHED, 1=VULNERABLE, 2=UNKNOWN

import os
import re
import sys
from pathlib import Path

try:
    from importlib.metadata import version, PackageNotFoundError
except Exception:
    print('UNKNOWN - importlib.metadata unavailable')
    sys.exit(2)

TARGET = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else None
PKG = 'pyfory'

def parse_ver(v):
    m = re.match(r'^(\d+)\.(\d+)\.(\d+)', str(v))
    if not m:
        return None
    return tuple(int(x) for x in m.groups())

def lt(a, b):
    return a < b

def scan_repo(root):
    findings = []
    if root is None or not root.exists():
        return findings
    exts = {'.py', '.pyw'}
    patterns = [
        re.compile(r'Fory\s*\([^\)]*strict\s*=\s*False', re.S),
        re.compile(r'Fory\s*\([^\)]*xlang\s*=\s*False', re.S),
        re.compile(r'DeserializationPolicy'),
        re.compile(r'pyfory\.(loads|deserialize)\s*\('),
        re.compile(r'\b(loads|deserialize)\s*\('),
    ]
    for p in root.rglob('*'):
        if not p.is_file() or p.suffix.lower() not in exts:
            continue
        try:
            text = p.read_text(encoding='utf-8', errors='ignore')
        except Exception:
            continue
        hits = []
        for pat in patterns:
            if pat.search(text):
                hits.append(pat.pattern)
        if hits:
            findings.append((str(p), hits))
    return findings

try:
    installed = version(PKG)
except PackageNotFoundError:
    print('UNKNOWN - pyfory not installed in this Python environment')
    sys.exit(2)
except Exception as e:
    print(f'UNKNOWN - failed to query package version: {e}')
    sys.exit(2)

installed_t = parse_ver(installed)
fixed_t = (1, 0, 0)

if installed_t is None:
    print(f'UNKNOWN - could not parse pyfory version: {installed}')
    sys.exit(2)

repo_findings = scan_repo(TARGET)
strict_false = False
python_native = False
policy_used = False
for _, hits in repo_findings:
    for h in hits:
        if 'strict\s*=\s*False' in h:
            strict_false = True
        if 'xlang\s*=\s*False' in h:
            python_native = True
        if 'DeserializationPolicy' in h:
            policy_used = True

if lt(installed_t, fixed_t):
    if strict_false and (python_native or policy_used):
        print(f'VULNERABLE - pyfory {installed} < 1.0.0 and code scan found risky usage (strict=False with Python-native/policy indicators)')
        if TARGET:
            for path, hits in repo_findings[:20]:
                print(f'  hit: {path} :: {", ".join(hits)}')
        sys.exit(1)
    else:
        print(f'UNKNOWN - pyfory {installed} is in the affected version range, but risky runtime usage was not confirmed from static scan alone')
        if TARGET and repo_findings:
            for path, hits in repo_findings[:20]:
                print(f'  context: {path} :: {", ".join(hits)}')
        sys.exit(2)
else:
    print(f'PATCHED - pyfory {installed} >= 1.0.0')
    sys.exit(0)