import subprocess
import sys
import tempfile
+import time
from datetime import datetime, timedelta, timezone
from enum import Enum, auto
from hashlib import pbkdf2_hmac
unpad_pkcs7,
)
from .compat import compat_b64decode, compat_cookiejar_Cookie
+from .dependencies import (
+ _SECRETSTORAGE_UNAVAILABLE_REASON,
+ secretstorage,
+ sqlite3,
+)
from .minicurses import MultilinePrinter, QuietMultilinePrinter
from .utils import Popen, YoutubeDLCookieJar, error_to_str, expand_path
-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 secretstorage
- SECRETSTORAGE_AVAILABLE = True
-except ImportError:
- SECRETSTORAGE_AVAILABLE = False
- SECRETSTORAGE_UNAVAILABLE_REASON = (
- 'as the `secretstorage` module is not installed. '
- 'Please install by running `python3 -m pip install secretstorage`.')
-except Exception as _err:
- SECRETSTORAGE_AVAILABLE = False
- SECRETSTORAGE_UNAVAILABLE_REASON = f'as the `secretstorage` module could not be initialized. {_err}'
-
-
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
return
except BaseException:
return
-
- printer = MultilinePrinter(file, preserve_output=False)
- printer.print = lambda message: printer.print_at_line(f'[Cookies] {message}', 0)
- return printer
+ return self.ProgressBar(file, preserve_output=False)
def _create_progress_bar(logger):
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)
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()
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:
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':
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)
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
unencrypted_cookies = 0
else:
failed_message = ''
logger.info(f'Extracted {len(jar)} cookies from {browser_name}{failed_message}')
- counts = decryptor.cookie_counts.copy()
+ counts = decryptor._cookie_counts.copy()
counts['unencrypted'] = unencrypted_cookies
logger.debug(f'cookie version breakdown: {counts}')
return jar
def _process_chrome_cookie(decryptor, host_key, name, value, encrypted_value, path, expires_utc, is_secure):
- host_key = host_key.decode('utf-8')
- name = name.decode('utf-8')
- value = value.decode('utf-8')
- path = path.decode('utf-8')
+ host_key = host_key.decode()
+ name = name.decode()
+ value = value.decode()
+ path = path.decode()
is_encrypted = not value and encrypted_value
if is_encrypted:
"""
def decrypt(self, encrypted_value):
- raise NotImplementedError
+ raise NotImplementedError('Must be implemented by sub classes')
@property
- def cookie_counts(self):
- raise NotImplementedError
+ def _cookie_counts(self):
+ raise NotImplementedError('Must be implemented by sub classes')
def get_cookie_decryptor(browser_root, browser_keyring_name, logger, *, keyring=None):
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):
# https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_linux.cc
return pbkdf2_sha1(password, salt=b'saltysalt', iterations=1, key_length=16)
- @property
- def cookie_counts(self):
- return self._cookie_counts
-
def decrypt(self, encrypted_value):
version = encrypted_value[:3]
ciphertext = encrypted_value[3:]
# https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_mac.mm
return pbkdf2_sha1(password, salt=b'saltysalt', iterations=1003, key_length=16)
- @property
- def cookie_counts(self):
- return self._cookie_counts
-
def decrypt(self, encrypted_value):
version = encrypted_value[:3]
ciphertext = encrypted_value[3:]
self._v10_key = _get_windows_v10_key(browser_root, logger)
self._cookie_counts = {'v10': 0, 'other': 0}
- @property
- def cookie_counts(self):
- return self._cookie_counts
-
def decrypt(self, encrypted_value):
version = encrypted_value[:3]
ciphertext = encrypted_value[3:]
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):
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(f'invalid skip of {num_bytes} bytes')
logger.warning('failed to read NetworkWallet')
return default_wallet
else:
- network_wallet = stdout.decode('utf-8').strip()
+ network_wallet = stdout.decode().strip()
logger.debug(f'NetworkWallet = "{network_wallet}"')
return network_wallet
except Exception as e:
stdout, stderr = proc.communicate_or_kill()
if proc.returncode != 0:
- logger.error('kwallet-query failed with return code {}. Please consult '
- 'the kwallet-query man page for details'.format(proc.returncode))
+ 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'):
def _get_gnome_keyring_password(browser_keyring_name, logger):
- if not SECRETSTORAGE_AVAILABLE:
- logger.error(f'secretstorage not available {SECRETSTORAGE_UNAVAILABLE_REASON}')
+ 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
def _decrypt_aes_cbc(ciphertext, key, logger, initialization_vector=b' ' * 16):
plaintext = unpad_pkcs7(aes_cbc_decrypt_bytes(ciphertext, key, initialization_vector))
try:
- return plaintext.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(f'PRAGMA table_info({table_name})').fetchall()
- return [row[1].decode('utf-8') for row in table_info]
+ return [row[1].decode() for row in table_info]
def _find_most_recently_used_file(root, filename, logger):