+import contextlib
import ctypes
import json
import os
import subprocess
import sys
import tempfile
+import time
from datetime import datetime, timedelta, timezone
+from enum import Enum, auto
from hashlib import pbkdf2_hmac
-from .aes import aes_cbc_decrypt_bytes, aes_gcm_decrypt_and_verify_bytes
-from .compat import (
- compat_b64decode,
- compat_cookiejar_Cookie,
+from .aes import (
+ aes_cbc_decrypt_bytes,
+ aes_gcm_decrypt_and_verify_bytes,
+ unpad_pkcs7,
)
-from .utils import (
- bug_reports_message,
- expand_path,
- Popen,
- YoutubeDLCookieJar,
+from .compat import compat_b64decode, compat_cookiejar_Cookie
+from .dependencies import (
+ _SECRETSTORAGE_UNAVAILABLE_REASON,
+ secretstorage,
+ sqlite3,
)
-
-try:
- import sqlite3
- SQLITE_AVAILABLE = True
-except ImportError:
- # although sqlite3 is part of the standard library, it is possible to compile python without
- # sqlite support. See: https://github.com/yt-dlp/yt-dlp/issues/544
- SQLITE_AVAILABLE = False
-
-
-try:
- import keyring
- KEYRING_AVAILABLE = True
- KEYRING_UNAVAILABLE_REASON = f'due to unknown reasons{bug_reports_message()}'
-except ImportError:
- KEYRING_AVAILABLE = False
- KEYRING_UNAVAILABLE_REASON = (
- 'as the `keyring` module is not installed. '
- 'Please install by running `python3 -m pip install keyring`. '
- 'Depending on your platform, additional packages may be required '
- 'to access the keyring; see https://pypi.org/project/keyring')
-except Exception as _err:
- KEYRING_AVAILABLE = False
- KEYRING_UNAVAILABLE_REASON = 'as the `keyring` module could not be initialized: %s' % _err
-
+from .minicurses import MultilinePrinter, QuietMultilinePrinter
+from .utils import Popen, YoutubeDLCookieJar, error_to_str, expand_path
CHROMIUM_BASED_BROWSERS = {'brave', 'chrome', 'chromium', 'edge', 'opera', 'vivaldi'}
SUPPORTED_BROWSERS = CHROMIUM_BASED_BROWSERS | {'firefox', 'safari'}
if self._ydl:
self._ydl.report_error(message)
+ class ProgressBar(MultilinePrinter):
+ _DELAY, _timer = 0.1, 0
+
+ def print(self, message):
+ if time.time() - self._timer > self._DELAY:
+ self.print_at_line(f'[Cookies] {message}', 0)
+ self._timer = time.time()
+
+ def progress_bar(self):
+ """Return a context manager with a print method. (Optional)"""
+ # Do not print to files/pipes, loggers, or when --no-progress is used
+ if not self._ydl or self._ydl.params.get('noprogress') or self._ydl.params.get('logger'):
+ return
+ file = self._ydl._out_files['error']
+ try:
+ if not file.isatty():
+ return
+ except BaseException:
+ return
+ return self.ProgressBar(file, preserve_output=False)
+
+
+def _create_progress_bar(logger):
+ if hasattr(logger, 'progress_bar'):
+ printer = logger.progress_bar()
+ if printer:
+ return printer
+ printer = QuietMultilinePrinter()
+ printer.print = lambda _: None
+ return printer
+
def load_cookies(cookie_file, browser_specification, ydl):
cookie_jars = []
if browser_specification is not None:
- browser_name, profile = _parse_browser_specification(*browser_specification)
- cookie_jars.append(extract_cookies_from_browser(browser_name, profile, YDLLogger(ydl)))
+ browser_name, profile, keyring = _parse_browser_specification(*browser_specification)
+ cookie_jars.append(extract_cookies_from_browser(browser_name, profile, YDLLogger(ydl), keyring=keyring))
if cookie_file is not None:
- cookie_file = expand_path(cookie_file)
+ is_filename = YoutubeDLCookieJar.is_path(cookie_file)
+ if is_filename:
+ cookie_file = expand_path(cookie_file)
+
jar = YoutubeDLCookieJar(cookie_file)
- if os.access(cookie_file, os.R_OK):
+ if not is_filename or os.access(cookie_file, os.R_OK):
jar.load(ignore_discard=True, ignore_expires=True)
cookie_jars.append(jar)
return _merge_cookie_jars(cookie_jars)
-def extract_cookies_from_browser(browser_name, profile=None, logger=YDLLogger()):
+def extract_cookies_from_browser(browser_name, profile=None, logger=YDLLogger(), *, keyring=None):
if browser_name == 'firefox':
return _extract_firefox_cookies(profile, logger)
elif browser_name == 'safari':
return _extract_safari_cookies(profile, logger)
elif browser_name in CHROMIUM_BASED_BROWSERS:
- return _extract_chrome_cookies(browser_name, profile, logger)
+ return _extract_chrome_cookies(browser_name, profile, keyring, logger)
else:
- raise ValueError('unknown browser: {}'.format(browser_name))
+ raise ValueError(f'unknown browser: {browser_name}')
def _extract_firefox_cookies(profile, logger):
logger.info('Extracting cookies from firefox')
- if not SQLITE_AVAILABLE:
+ if not sqlite3:
logger.warning('Cannot extract cookies from firefox without sqlite3 support. '
'Please use a python interpreter compiled with sqlite3 support')
return YoutubeDLCookieJar()
else:
search_root = os.path.join(_firefox_browser_dir(), profile)
- cookie_database_path = _find_most_recently_used_file(search_root, 'cookies.sqlite')
+ cookie_database_path = _find_most_recently_used_file(search_root, 'cookies.sqlite', logger)
if cookie_database_path is None:
- raise FileNotFoundError('could not find firefox cookies database in {}'.format(search_root))
- logger.debug('Extracting cookies from: "{}"'.format(cookie_database_path))
+ raise FileNotFoundError(f'could not find firefox cookies database in {search_root}')
+ logger.debug(f'Extracting cookies from: "{cookie_database_path}"')
with tempfile.TemporaryDirectory(prefix='yt_dlp') as tmpdir:
cursor = None
cursor = _open_database_copy(cookie_database_path, tmpdir)
cursor.execute('SELECT host, name, value, path, expiry, isSecure FROM moz_cookies')
jar = YoutubeDLCookieJar()
- for host, name, value, path, expiry, is_secure in cursor.fetchall():
- cookie = compat_cookiejar_Cookie(
- version=0, name=name, value=value, port=None, port_specified=False,
- domain=host, domain_specified=bool(host), domain_initial_dot=host.startswith('.'),
- path=path, path_specified=bool(path), secure=is_secure, expires=expiry, discard=False,
- comment=None, comment_url=None, rest={})
- jar.set_cookie(cookie)
- logger.info('Extracted {} cookies from firefox'.format(len(jar)))
+ with _create_progress_bar(logger) as progress_bar:
+ table = cursor.fetchall()
+ total_cookie_count = len(table)
+ for i, (host, name, value, path, expiry, is_secure) in enumerate(table):
+ progress_bar.print(f'Loading cookie {i: 6d}/{total_cookie_count: 6d}')
+ cookie = compat_cookiejar_Cookie(
+ version=0, name=name, value=value, port=None, port_specified=False,
+ domain=host, domain_specified=bool(host), domain_initial_dot=host.startswith('.'),
+ path=path, path_specified=bool(path), secure=is_secure, expires=expiry, discard=False,
+ comment=None, comment_url=None, rest={})
+ jar.set_cookie(cookie)
+ logger.info(f'Extracted {len(jar)} cookies from firefox')
return jar
finally:
if cursor is not None:
if sys.platform in ('linux', 'linux2'):
return os.path.expanduser('~/.mozilla/firefox')
elif sys.platform == 'win32':
- return os.path.expandvars(r'%APPDATA%\Mozilla\Firefox\Profiles')
+ return os.path.expandvars(R'%APPDATA%\Mozilla\Firefox\Profiles')
elif sys.platform == 'darwin':
return os.path.expanduser('~/Library/Application Support/Firefox')
else:
- raise ValueError('unsupported platform: {}'.format(sys.platform))
+ raise ValueError(f'unsupported platform: {sys.platform}')
def _get_chromium_based_browser_settings(browser_name):
appdata_local = os.path.expandvars('%LOCALAPPDATA%')
appdata_roaming = os.path.expandvars('%APPDATA%')
browser_dir = {
- 'brave': os.path.join(appdata_local, r'BraveSoftware\Brave-Browser\User Data'),
- 'chrome': os.path.join(appdata_local, r'Google\Chrome\User Data'),
- 'chromium': os.path.join(appdata_local, r'Chromium\User Data'),
- 'edge': os.path.join(appdata_local, r'Microsoft\Edge\User Data'),
- 'opera': os.path.join(appdata_roaming, r'Opera Software\Opera Stable'),
- 'vivaldi': os.path.join(appdata_local, r'Vivaldi\User Data'),
+ 'brave': os.path.join(appdata_local, R'BraveSoftware\Brave-Browser\User Data'),
+ 'chrome': os.path.join(appdata_local, R'Google\Chrome\User Data'),
+ 'chromium': os.path.join(appdata_local, R'Chromium\User Data'),
+ 'edge': os.path.join(appdata_local, R'Microsoft\Edge\User Data'),
+ 'opera': os.path.join(appdata_roaming, R'Opera Software\Opera Stable'),
+ 'vivaldi': os.path.join(appdata_local, R'Vivaldi\User Data'),
}[browser_name]
elif sys.platform == 'darwin':
}[browser_name]
else:
- raise ValueError('unsupported platform: {}'.format(sys.platform))
+ raise ValueError(f'unsupported platform: {sys.platform}')
# Linux keyring names can be determined by snooping on dbus while opening the browser in KDE:
# dbus-monitor "interface='org.kde.KWallet'" "type=method_return"
}
-def _extract_chrome_cookies(browser_name, profile, logger):
- logger.info('Extracting cookies from {}'.format(browser_name))
+def _extract_chrome_cookies(browser_name, profile, keyring, logger):
+ logger.info(f'Extracting cookies from {browser_name}')
- if not SQLITE_AVAILABLE:
- logger.warning(('Cannot extract cookies from {} without sqlite3 support. '
- 'Please use a python interpreter compiled with sqlite3 support').format(browser_name))
+ if not sqlite3:
+ logger.warning(f'Cannot extract cookies from {browser_name} without sqlite3 support. '
+ 'Please use a python interpreter compiled with sqlite3 support')
return YoutubeDLCookieJar()
config = _get_chromium_based_browser_settings(browser_name)
if config['supports_profiles']:
search_root = os.path.join(config['browser_dir'], profile)
else:
- logger.error('{} does not support profiles'.format(browser_name))
+ logger.error(f'{browser_name} does not support profiles')
search_root = config['browser_dir']
- cookie_database_path = _find_most_recently_used_file(search_root, 'Cookies')
+ cookie_database_path = _find_most_recently_used_file(search_root, 'Cookies', logger)
if cookie_database_path is None:
- raise FileNotFoundError('could not find {} cookies database in "{}"'.format(browser_name, search_root))
- logger.debug('Extracting cookies from: "{}"'.format(cookie_database_path))
+ raise FileNotFoundError(f'could not find {browser_name} cookies database in "{search_root}"')
+ logger.debug(f'Extracting cookies from: "{cookie_database_path}"')
- decryptor = get_cookie_decryptor(config['browser_dir'], config['keyring_name'], logger)
+ decryptor = get_cookie_decryptor(config['browser_dir'], config['keyring_name'], logger, keyring=keyring)
with tempfile.TemporaryDirectory(prefix='yt_dlp') as tmpdir:
cursor = None
cursor.connection.text_factory = bytes
column_names = _get_column_names(cursor, 'cookies')
secure_column = 'is_secure' if 'is_secure' in column_names else 'secure'
- cursor.execute('SELECT host_key, name, value, encrypted_value, path, '
- 'expires_utc, {} FROM cookies'.format(secure_column))
+ cursor.execute(f'SELECT host_key, name, value, encrypted_value, path, expires_utc, {secure_column} FROM cookies')
jar = YoutubeDLCookieJar()
failed_cookies = 0
- for host_key, name, value, encrypted_value, path, expires_utc, is_secure in cursor.fetchall():
- host_key = host_key.decode('utf-8')
- name = name.decode('utf-8')
- value = value.decode('utf-8')
- path = path.decode('utf-8')
-
- if not value and encrypted_value:
- value = decryptor.decrypt(encrypted_value)
- if value is None:
+ unencrypted_cookies = 0
+ with _create_progress_bar(logger) as progress_bar:
+ table = cursor.fetchall()
+ total_cookie_count = len(table)
+ for i, line in enumerate(table):
+ progress_bar.print(f'Loading cookie {i: 6d}/{total_cookie_count: 6d}')
+ is_encrypted, cookie = _process_chrome_cookie(decryptor, *line)
+ if not cookie:
failed_cookies += 1
continue
-
- cookie = compat_cookiejar_Cookie(
- version=0, name=name, value=value, port=None, port_specified=False,
- domain=host_key, domain_specified=bool(host_key), domain_initial_dot=host_key.startswith('.'),
- path=path, path_specified=bool(path), secure=is_secure, expires=expires_utc, discard=False,
- comment=None, comment_url=None, rest={})
- jar.set_cookie(cookie)
+ elif not is_encrypted:
+ unencrypted_cookies += 1
+ jar.set_cookie(cookie)
if failed_cookies > 0:
- failed_message = ' ({} could not be decrypted)'.format(failed_cookies)
+ failed_message = f' ({failed_cookies} could not be decrypted)'
else:
failed_message = ''
- logger.info('Extracted {} cookies from {}{}'.format(len(jar), browser_name, failed_message))
+ logger.info(f'Extracted {len(jar)} cookies from {browser_name}{failed_message}')
+ counts = decryptor._cookie_counts.copy()
+ counts['unencrypted'] = unencrypted_cookies
+ logger.debug(f'cookie version breakdown: {counts}')
return jar
finally:
if cursor is not None:
cursor.connection.close()
+def _process_chrome_cookie(decryptor, host_key, name, value, encrypted_value, path, expires_utc, is_secure):
+ host_key = host_key.decode()
+ name = name.decode()
+ value = value.decode()
+ path = path.decode()
+ is_encrypted = not value and encrypted_value
+
+ if is_encrypted:
+ value = decryptor.decrypt(encrypted_value)
+ if value is None:
+ return is_encrypted, None
+
+ return is_encrypted, compat_cookiejar_Cookie(
+ version=0, name=name, value=value, port=None, port_specified=False,
+ domain=host_key, domain_specified=bool(host_key), domain_initial_dot=host_key.startswith('.'),
+ path=path, path_specified=bool(path), secure=is_secure, expires=expires_utc, discard=False,
+ comment=None, comment_url=None, rest={})
+
+
class ChromeCookieDecryptor:
"""
Overview:
"""
def decrypt(self, encrypted_value):
- raise NotImplementedError
+ raise NotImplementedError('Must be implemented by sub classes')
+
+ @property
+ def _cookie_counts(self):
+ raise NotImplementedError('Must be implemented by sub classes')
-def get_cookie_decryptor(browser_root, browser_keyring_name, logger):
+def get_cookie_decryptor(browser_root, browser_keyring_name, logger, *, keyring=None):
if sys.platform in ('linux', 'linux2'):
- return LinuxChromeCookieDecryptor(browser_keyring_name, logger)
+ return LinuxChromeCookieDecryptor(browser_keyring_name, logger, keyring=keyring)
elif sys.platform == 'darwin':
return MacChromeCookieDecryptor(browser_keyring_name, logger)
elif sys.platform == 'win32':
return WindowsChromeCookieDecryptor(browser_root, logger)
else:
- raise NotImplementedError('Chrome cookie decryption is not supported '
- 'on this platform: {}'.format(sys.platform))
+ raise NotImplementedError(f'Chrome cookie decryption is not supported on this platform: {sys.platform}')
class LinuxChromeCookieDecryptor(ChromeCookieDecryptor):
- def __init__(self, browser_keyring_name, logger):
+ def __init__(self, browser_keyring_name, logger, *, keyring=None):
self._logger = logger
self._v10_key = self.derive_key(b'peanuts')
- if KEYRING_AVAILABLE:
- self._v11_key = self.derive_key(_get_linux_keyring_password(browser_keyring_name))
- else:
- self._v11_key = None
+ password = _get_linux_keyring_password(browser_keyring_name, keyring, logger)
+ self._v11_key = None if password is None else self.derive_key(password)
+ self._cookie_counts = {'v10': 0, 'v11': 0, 'other': 0}
@staticmethod
def derive_key(password):
ciphertext = encrypted_value[3:]
if version == b'v10':
+ self._cookie_counts['v10'] += 1
return _decrypt_aes_cbc(ciphertext, self._v10_key, self._logger)
elif version == b'v11':
+ self._cookie_counts['v11'] += 1
if self._v11_key is None:
- self._logger.warning(f'cannot decrypt cookie {KEYRING_UNAVAILABLE_REASON}', only_once=True)
+ self._logger.warning('cannot decrypt v11 cookies: no key found', only_once=True)
return None
return _decrypt_aes_cbc(ciphertext, self._v11_key, self._logger)
else:
+ self._cookie_counts['other'] += 1
return None
self._logger = logger
password = _get_mac_keyring_password(browser_keyring_name, logger)
self._v10_key = None if password is None else self.derive_key(password)
+ self._cookie_counts = {'v10': 0, 'other': 0}
@staticmethod
def derive_key(password):
ciphertext = encrypted_value[3:]
if version == b'v10':
+ self._cookie_counts['v10'] += 1
if self._v10_key is None:
self._logger.warning('cannot decrypt v10 cookies: no key found', only_once=True)
return None
return _decrypt_aes_cbc(ciphertext, self._v10_key, self._logger)
else:
+ self._cookie_counts['other'] += 1
# other prefixes are considered 'old data' which were stored as plaintext
# https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_mac.mm
return encrypted_value
def __init__(self, browser_root, logger):
self._logger = logger
self._v10_key = _get_windows_v10_key(browser_root, logger)
+ self._cookie_counts = {'v10': 0, 'other': 0}
def decrypt(self, encrypted_value):
version = encrypted_value[:3]
ciphertext = encrypted_value[3:]
if version == b'v10':
+ self._cookie_counts['v10'] += 1
if self._v10_key is None:
self._logger.warning('cannot decrypt v10 cookies: no key found', only_once=True)
return None
return _decrypt_aes_gcm(ciphertext, self._v10_key, nonce, authentication_tag, self._logger)
else:
+ self._cookie_counts['other'] += 1
# any other prefix means the data is DPAPI encrypted
# https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_win.cc
- return _decrypt_windows_dpapi(encrypted_value, self._logger).decode('utf-8')
+ return _decrypt_windows_dpapi(encrypted_value, self._logger).decode()
def _extract_safari_cookies(profile, logger):
if profile is not None:
logger.error('safari does not support profiles')
if sys.platform != 'darwin':
- raise ValueError('unsupported platform: {}'.format(sys.platform))
+ raise ValueError(f'unsupported platform: {sys.platform}')
cookies_path = os.path.expanduser('~/Library/Cookies/Cookies.binarycookies')
if not os.path.isfile(cookies_path):
- raise FileNotFoundError('could not find safari cookies database')
+ logger.debug('Trying secondary cookie location')
+ cookies_path = os.path.expanduser('~/Library/Containers/com.apple.Safari/Data/Library/Cookies/Cookies.binarycookies')
+ if not os.path.isfile(cookies_path):
+ raise FileNotFoundError('could not find safari cookies database')
with open(cookies_path, 'rb') as f:
cookies_data = f.read()
jar = parse_safari_cookies(cookies_data, logger=logger)
- logger.info('Extracted {} cookies from safari'.format(len(jar)))
+ logger.info(f'Extracted {len(jar)} cookies from safari')
return jar
def read_bytes(self, num_bytes):
if num_bytes < 0:
- raise ParserError('invalid read of {} bytes'.format(num_bytes))
+ raise ParserError(f'invalid read of {num_bytes} bytes')
end = self.cursor + num_bytes
if end > len(self._data):
raise ParserError('reached end of input')
def expect_bytes(self, expected_value, message):
value = self.read_bytes(len(expected_value))
if value != expected_value:
- raise ParserError('unexpected value: {} != {} ({})'.format(value, expected_value, message))
+ raise ParserError(f'unexpected value: {value} != {expected_value} ({message})')
def read_uint(self, big_endian=False):
data_format = '>I' if big_endian else '<I'
while True:
c = self.read_bytes(1)
if c == b'\x00':
- return b''.join(buffer).decode('utf-8')
+ return b''.join(buffer).decode()
else:
buffer.append(c)
def skip(self, num_bytes, description='unknown'):
if num_bytes > 0:
- self._logger.debug('skipping {} bytes ({}): {}'.format(
- num_bytes, description, self.read_bytes(num_bytes)))
+ self._logger.debug(f'skipping {num_bytes} bytes ({description}): {self.read_bytes(num_bytes)!r}')
elif num_bytes < 0:
- raise ParserError('invalid skip of {} bytes'.format(num_bytes))
+ raise ParserError(f'invalid skip of {num_bytes} bytes')
def skip_to(self, offset, description='unknown'):
self.skip(offset - self.cursor, description)
number_of_cookies = p.read_uint()
record_offsets = [p.read_uint() for _ in range(number_of_cookies)]
if number_of_cookies == 0:
- logger.debug('a cookies page of size {} has no cookies'.format(len(data)))
+ logger.debug(f'a cookies page of size {len(data)} has no cookies')
return
p.skip_to(record_offsets[0], 'unknown page header field')
- for record_offset in record_offsets:
- p.skip_to(record_offset, 'space between records')
- record_length = _parse_safari_cookies_record(data[record_offset:], jar, logger)
- p.read_bytes(record_length)
+ with _create_progress_bar(logger) as progress_bar:
+ for i, record_offset in enumerate(record_offsets):
+ progress_bar.print(f'Loading cookie {i: 6d}/{number_of_cookies: 6d}')
+ p.skip_to(record_offset, 'space between records')
+ record_length = _parse_safari_cookies_record(data[record_offset:], jar, logger)
+ p.read_bytes(record_length)
p.skip_to_end('space in between pages')
return jar
-def _get_linux_keyring_password(browser_keyring_name):
- password = keyring.get_password('{} Keys'.format(browser_keyring_name),
- '{} Safe Storage'.format(browser_keyring_name))
- if password is None:
- # this sometimes occurs in KDE because chrome does not check hasEntry and instead
- # just tries to read the value (which kwallet returns "") whereas keyring checks hasEntry
- # to verify this:
- # dbus-monitor "interface='org.kde.KWallet'" "type=method_return"
- # while starting chrome.
- # this may be a bug as the intended behaviour is to generate a random password and store
- # it, but that doesn't matter here.
- password = ''
- return password.encode('utf-8')
+class _LinuxDesktopEnvironment(Enum):
+ """
+ https://chromium.googlesource.com/chromium/src/+/refs/heads/main/base/nix/xdg_util.h
+ DesktopEnvironment
+ """
+ OTHER = auto()
+ CINNAMON = auto()
+ GNOME = auto()
+ KDE = auto()
+ PANTHEON = auto()
+ UNITY = auto()
+ XFCE = auto()
-def _get_mac_keyring_password(browser_keyring_name, logger):
- if KEYRING_AVAILABLE:
- logger.debug('using keyring to obtain password')
- password = keyring.get_password('{} Safe Storage'.format(browser_keyring_name), browser_keyring_name)
- return password.encode('utf-8')
+class _LinuxKeyring(Enum):
+ """
+ https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/key_storage_util_linux.h
+ SelectedLinuxBackend
+ """
+ KWALLET = auto()
+ GNOMEKEYRING = auto()
+ BASICTEXT = auto()
+
+
+SUPPORTED_KEYRINGS = _LinuxKeyring.__members__.keys()
+
+
+def _get_linux_desktop_environment(env):
+ """
+ https://chromium.googlesource.com/chromium/src/+/refs/heads/main/base/nix/xdg_util.cc
+ GetDesktopEnvironment
+ """
+ xdg_current_desktop = env.get('XDG_CURRENT_DESKTOP', None)
+ desktop_session = env.get('DESKTOP_SESSION', None)
+ if xdg_current_desktop is not None:
+ xdg_current_desktop = xdg_current_desktop.split(':')[0].strip()
+
+ if xdg_current_desktop == 'Unity':
+ if desktop_session is not None and 'gnome-fallback' in desktop_session:
+ return _LinuxDesktopEnvironment.GNOME
+ else:
+ return _LinuxDesktopEnvironment.UNITY
+ elif xdg_current_desktop == 'GNOME':
+ return _LinuxDesktopEnvironment.GNOME
+ elif xdg_current_desktop == 'X-Cinnamon':
+ return _LinuxDesktopEnvironment.CINNAMON
+ elif xdg_current_desktop == 'KDE':
+ return _LinuxDesktopEnvironment.KDE
+ elif xdg_current_desktop == 'Pantheon':
+ return _LinuxDesktopEnvironment.PANTHEON
+ elif xdg_current_desktop == 'XFCE':
+ return _LinuxDesktopEnvironment.XFCE
+ elif desktop_session is not None:
+ if desktop_session in ('mate', 'gnome'):
+ return _LinuxDesktopEnvironment.GNOME
+ elif 'kde' in desktop_session:
+ return _LinuxDesktopEnvironment.KDE
+ elif 'xfce' in desktop_session:
+ return _LinuxDesktopEnvironment.XFCE
+ else:
+ if 'GNOME_DESKTOP_SESSION_ID' in env:
+ return _LinuxDesktopEnvironment.GNOME
+ elif 'KDE_FULL_SESSION' in env:
+ return _LinuxDesktopEnvironment.KDE
+ return _LinuxDesktopEnvironment.OTHER
+
+
+def _choose_linux_keyring(logger):
+ """
+ https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/key_storage_util_linux.cc
+ SelectBackend
+ """
+ desktop_environment = _get_linux_desktop_environment(os.environ)
+ logger.debug(f'detected desktop environment: {desktop_environment.name}')
+ if desktop_environment == _LinuxDesktopEnvironment.KDE:
+ linux_keyring = _LinuxKeyring.KWALLET
+ elif desktop_environment == _LinuxDesktopEnvironment.OTHER:
+ linux_keyring = _LinuxKeyring.BASICTEXT
else:
- logger.debug('using find-generic-password to obtain password')
+ linux_keyring = _LinuxKeyring.GNOMEKEYRING
+ return linux_keyring
+
+
+def _get_kwallet_network_wallet(logger):
+ """ The name of the wallet used to store network passwords.
+
+ https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/kwallet_dbus.cc
+ KWalletDBus::NetworkWallet
+ which does a dbus call to the following function:
+ https://api.kde.org/frameworks/kwallet/html/classKWallet_1_1Wallet.html
+ Wallet::NetworkWallet
+ """
+ default_wallet = 'kdewallet'
+ try:
+ proc = Popen([
+ 'dbus-send', '--session', '--print-reply=literal',
+ '--dest=org.kde.kwalletd5',
+ '/modules/kwalletd5',
+ 'org.kde.KWallet.networkWallet'
+ ], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
+
+ stdout, stderr = proc.communicate_or_kill()
+ if proc.returncode != 0:
+ logger.warning('failed to read NetworkWallet')
+ return default_wallet
+ else:
+ network_wallet = stdout.decode().strip()
+ logger.debug(f'NetworkWallet = "{network_wallet}"')
+ return network_wallet
+ except Exception as e:
+ logger.warning(f'exception while obtaining NetworkWallet: {e}')
+ return default_wallet
+
+
+def _get_kwallet_password(browser_keyring_name, logger):
+ logger.debug('using kwallet-query to obtain password from kwallet')
+
+ if shutil.which('kwallet-query') is None:
+ logger.error('kwallet-query command not found. KWallet and kwallet-query '
+ 'must be installed to read from KWallet. kwallet-query should be'
+ 'included in the kwallet package for your distribution')
+ return b''
+
+ network_wallet = _get_kwallet_network_wallet(logger)
+
+ try:
+ proc = Popen([
+ 'kwallet-query',
+ '--read-password', f'{browser_keyring_name} Safe Storage',
+ '--folder', f'{browser_keyring_name} Keys',
+ network_wallet
+ ], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
+
+ stdout, stderr = proc.communicate_or_kill()
+ if proc.returncode != 0:
+ logger.error(f'kwallet-query failed with return code {proc.returncode}. Please consult '
+ 'the kwallet-query man page for details')
+ return b''
+ else:
+ if stdout.lower().startswith(b'failed to read'):
+ logger.debug('failed to read password from kwallet. Using empty string instead')
+ # this sometimes occurs in KDE because chrome does not check hasEntry and instead
+ # just tries to read the value (which kwallet returns "") whereas kwallet-query
+ # checks hasEntry. To verify this:
+ # dbus-monitor "interface='org.kde.KWallet'" "type=method_return"
+ # while starting chrome.
+ # this may be a bug as the intended behaviour is to generate a random password and store
+ # it, but that doesn't matter here.
+ return b''
+ else:
+ logger.debug('password found')
+ if stdout[-1:] == b'\n':
+ stdout = stdout[:-1]
+ return stdout
+ except Exception as e:
+ logger.warning(f'exception running kwallet-query: {error_to_str(e)}')
+ return b''
+
+
+def _get_gnome_keyring_password(browser_keyring_name, logger):
+ if not secretstorage:
+ logger.error(f'secretstorage not available {_SECRETSTORAGE_UNAVAILABLE_REASON}')
+ return b''
+ # the Gnome keyring does not seem to organise keys in the same way as KWallet,
+ # using `dbus-monitor` during startup, it can be observed that chromium lists all keys
+ # and presumably searches for its key in the list. It appears that we must do the same.
+ # https://github.com/jaraco/keyring/issues/556
+ with contextlib.closing(secretstorage.dbus_init()) as con:
+ col = secretstorage.get_default_collection(con)
+ for item in col.get_all_items():
+ if item.get_label() == f'{browser_keyring_name} Safe Storage':
+ return item.get_secret()
+ else:
+ logger.error('failed to read from keyring')
+ return b''
+
+
+def _get_linux_keyring_password(browser_keyring_name, keyring, logger):
+ # note: chrome/chromium can be run with the following flags to determine which keyring backend
+ # it has chosen to use
+ # chromium --enable-logging=stderr --v=1 2>&1 | grep key_storage_
+ # Chromium supports a flag: --password-store=<basic|gnome|kwallet> so the automatic detection
+ # will not be sufficient in all cases.
+
+ keyring = _LinuxKeyring[keyring] if keyring else _choose_linux_keyring(logger)
+ logger.debug(f'Chosen keyring: {keyring.name}')
+
+ if keyring == _LinuxKeyring.KWALLET:
+ return _get_kwallet_password(browser_keyring_name, logger)
+ elif keyring == _LinuxKeyring.GNOMEKEYRING:
+ return _get_gnome_keyring_password(browser_keyring_name, logger)
+ elif keyring == _LinuxKeyring.BASICTEXT:
+ # when basic text is chosen, all cookies are stored as v10 (so no keyring password is required)
+ return None
+ assert False, f'Unknown keyring {keyring}'
+
+
+def _get_mac_keyring_password(browser_keyring_name, logger):
+ logger.debug('using find-generic-password to obtain password from OSX keychain')
+ try:
proc = Popen(
['security', 'find-generic-password',
'-w', # write password to stdout
'-a', browser_keyring_name, # match 'account'
- '-s', '{} Safe Storage'.format(browser_keyring_name)], # match 'service'
+ '-s', f'{browser_keyring_name} Safe Storage'], # match 'service'
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
- try:
- stdout, stderr = proc.communicate_or_kill()
- if stdout[-1:] == b'\n':
- stdout = stdout[:-1]
- return stdout
- except BaseException as e:
- logger.warning(f'exception running find-generic-password: {type(e).__name__}({e})')
- return None
+
+ stdout, stderr = proc.communicate_or_kill()
+ if stdout[-1:] == b'\n':
+ stdout = stdout[:-1]
+ return stdout
+ except Exception as e:
+ logger.warning(f'exception running find-generic-password: {error_to_str(e)}')
+ return None
def _get_windows_v10_key(browser_root, logger):
- path = _find_most_recently_used_file(browser_root, 'Local State')
+ path = _find_most_recently_used_file(browser_root, 'Local State', logger)
if path is None:
logger.error('could not find local state file')
return None
- with open(path, 'r', encoding='utf8') as f:
+ logger.debug(f'Found local state file at "{path}"')
+ with open(path, encoding='utf8') as f:
data = json.load(f)
try:
base64_key = data['os_crypt']['encrypted_key']
def _decrypt_aes_cbc(ciphertext, key, logger, initialization_vector=b' ' * 16):
- plaintext = aes_cbc_decrypt_bytes(ciphertext, key, initialization_vector)
- padding_length = plaintext[-1]
+ plaintext = unpad_pkcs7(aes_cbc_decrypt_bytes(ciphertext, key, initialization_vector))
try:
- return plaintext[:-padding_length].decode('utf-8')
+ return plaintext.decode()
except UnicodeDecodeError:
logger.warning('failed to decrypt cookie (AES-CBC) because UTF-8 decoding failed. Possibly the key is wrong?', only_once=True)
return None
return None
try:
- return plaintext.decode('utf-8')
+ return plaintext.decode()
except UnicodeDecodeError:
logger.warning('failed to decrypt cookie (AES-GCM) because UTF-8 decoding failed. Possibly the key is wrong?', only_once=True)
return None
def _get_column_names(cursor, table_name):
- table_info = cursor.execute('PRAGMA table_info({})'.format(table_name)).fetchall()
- return [row[1].decode('utf-8') for row in table_info]
+ table_info = cursor.execute(f'PRAGMA table_info({table_name})').fetchall()
+ return [row[1].decode() for row in table_info]
-def _find_most_recently_used_file(root, filename):
+def _find_most_recently_used_file(root, filename, logger):
# if there are multiple browser profiles, take the most recently used one
- paths = []
- for root, dirs, files in os.walk(root):
- for file in files:
- if file == filename:
- paths.append(os.path.join(root, file))
+ i, paths = 0, []
+ with _create_progress_bar(logger) as progress_bar:
+ for curr_root, dirs, files in os.walk(root):
+ for file in files:
+ i += 1
+ progress_bar.print(f'Searching for "{filename}": {i: 6d} files searched')
+ if file == filename:
+ paths.append(os.path.join(curr_root, file))
return None if not paths else max(paths, key=lambda path: os.lstat(path).st_mtime)
return os.path.sep in value
-def _parse_browser_specification(browser_name, profile=None):
- browser_name = browser_name.lower()
+def _parse_browser_specification(browser_name, profile=None, keyring=None):
if browser_name not in SUPPORTED_BROWSERS:
raise ValueError(f'unsupported browser: "{browser_name}"')
+ if keyring not in (None, *SUPPORTED_KEYRINGS):
+ raise ValueError(f'unsupported keyring: "{keyring}"')
if profile is not None and _is_path(profile):
profile = os.path.expanduser(profile)
- return browser_name, profile
+ return browser_name, profile, keyring
projects / yt-dlp.git / blobdiff