cookies.py 52.5 KB
Newer Older
1
import base64
2
import collections
3
import contextlib
4
import http.cookiejar
5
import http.cookies
6
import io
7 8
import json
import os
9
import re
10 11 12 13 14
import shutil
import struct
import subprocess
import sys
import tempfile
P
pukkandan 已提交
15
import time
16
import urllib.request
17
from datetime import datetime, timedelta, timezone
18
from enum import Enum, auto
19 20
from hashlib import pbkdf2_hmac

P
pukkandan 已提交
21 22 23 24 25
from .aes import (
    aes_cbc_decrypt_bytes,
    aes_gcm_decrypt_and_verify_bytes,
    unpad_pkcs7,
)
26
from .compat import functools
27 28 29 30 31
from .dependencies import (
    _SECRETSTORAGE_UNAVAILABLE_REASON,
    secretstorage,
    sqlite3,
)
32
from .minicurses import MultilinePrinter, QuietMultilinePrinter
P
pukkandan 已提交
33 34 35
from .utils import (
    Popen,
    error_to_str,
36
    escape_url,
P
pukkandan 已提交
37
    expand_path,
38
    is_path_like,
39 40
    sanitize_url,
    str_or_none,
P
pukkandan 已提交
41
    try_call,
42
    write_string,
P
pukkandan 已提交
43
)
44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68

CHROMIUM_BASED_BROWSERS = {'brave', 'chrome', 'chromium', 'edge', 'opera', 'vivaldi'}
SUPPORTED_BROWSERS = CHROMIUM_BASED_BROWSERS | {'firefox', 'safari'}


class YDLLogger:
    def __init__(self, ydl=None):
        self._ydl = ydl

    def debug(self, message):
        if self._ydl:
            self._ydl.write_debug(message)

    def info(self, message):
        if self._ydl:
            self._ydl.to_screen(f'[Cookies] {message}')

    def warning(self, message, only_once=False):
        if self._ydl:
            self._ydl.report_warning(message, only_once)

    def error(self, message):
        if self._ydl:
            self._ydl.report_error(message)

P
pukkandan 已提交
69 70 71 72 73 74 75 76
    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()

77 78 79 80 81
    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
P
pukkandan 已提交
82
        file = self._ydl._out_files.error
83 84 85 86 87
        try:
            if not file.isatty():
                return
        except BaseException:
            return
P
pukkandan 已提交
88
        return self.ProgressBar(file, preserve_output=False)
89 90 91 92 93 94 95 96 97 98 99


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

100 101 102 103

def load_cookies(cookie_file, browser_specification, ydl):
    cookie_jars = []
    if browser_specification is not None:
104 105 106
        browser_name, profile, keyring, container = _parse_browser_specification(*browser_specification)
        cookie_jars.append(
            extract_cookies_from_browser(browser_name, profile, YDLLogger(ydl), keyring=keyring, container=container))
107 108

    if cookie_file is not None:
109
        is_filename = is_path_like(cookie_file)
110 111 112
        if is_filename:
            cookie_file = expand_path(cookie_file)

113
        jar = YoutubeDLCookieJar(cookie_file)
114
        if not is_filename or os.access(cookie_file, os.R_OK):
115 116 117 118 119 120
            jar.load(ignore_discard=True, ignore_expires=True)
        cookie_jars.append(jar)

    return _merge_cookie_jars(cookie_jars)


121
def extract_cookies_from_browser(browser_name, profile=None, logger=YDLLogger(), *, keyring=None, container=None):
122
    if browser_name == 'firefox':
123
        return _extract_firefox_cookies(profile, container, logger)
124 125 126
    elif browser_name == 'safari':
        return _extract_safari_cookies(profile, logger)
    elif browser_name in CHROMIUM_BASED_BROWSERS:
127
        return _extract_chrome_cookies(browser_name, profile, keyring, logger)
128
    else:
P
pukkandan 已提交
129
        raise ValueError(f'unknown browser: {browser_name}')
130 131


132
def _extract_firefox_cookies(profile, container, logger):
133
    logger.info('Extracting cookies from firefox')
134
    if not sqlite3:
135 136 137
        logger.warning('Cannot extract cookies from firefox without sqlite3 support. '
                       'Please use a python interpreter compiled with sqlite3 support')
        return YoutubeDLCookieJar()
138 139 140 141 142 143 144 145

    if profile is None:
        search_root = _firefox_browser_dir()
    elif _is_path(profile):
        search_root = profile
    else:
        search_root = os.path.join(_firefox_browser_dir(), profile)

146 147 148 149 150
    cookie_database_path = _find_most_recently_used_file(search_root, 'cookies.sqlite', logger)
    if cookie_database_path is None:
        raise FileNotFoundError(f'could not find firefox cookies database in {search_root}')
    logger.debug(f'Extracting cookies from: "{cookie_database_path}"')

151
    container_id = None
152 153
    if container not in (None, 'none'):
        containers_path = os.path.join(os.path.dirname(cookie_database_path), 'containers.json')
154 155
        if not os.path.isfile(containers_path) or not os.access(containers_path, os.R_OK):
            raise FileNotFoundError(f'could not read containers.json in {search_root}')
P
pukkandan 已提交
156
        with open(containers_path) as containers:
157 158 159 160 161 162 163 164
            identities = json.load(containers).get('identities', [])
        container_id = next((context.get('userContextId') for context in identities if container in (
            context.get('name'),
            try_call(lambda: re.fullmatch(r'userContext([^\.]+)\.label', context['l10nID']).group())
        )), None)
        if not isinstance(container_id, int):
            raise ValueError(f'could not find firefox container "{container}" in containers.json')

165
    with tempfile.TemporaryDirectory(prefix='yt_dlp') as tmpdir:
166 167 168
        cursor = None
        try:
            cursor = _open_database_copy(cookie_database_path, tmpdir)
169 170 171 172
            if isinstance(container_id, int):
                logger.debug(
                    f'Only loading cookies from firefox container "{container}", ID {container_id}')
                cursor.execute(
173 174 175 176 177 178 179
                    'SELECT host, name, value, path, expiry, isSecure FROM moz_cookies WHERE originAttributes LIKE ? OR originAttributes LIKE ?',
                    (f'%userContextId={container_id}', f'%userContextId={container_id}&%'))
            elif container == 'none':
                logger.debug('Only loading cookies not belonging to any container')
                cursor.execute(
                    'SELECT host, name, value, path, expiry, isSecure FROM moz_cookies WHERE NOT INSTR(originAttributes,"userContextId=")')
            else:
180
                cursor.execute('SELECT host, name, value, path, expiry, isSecure FROM moz_cookies')
181
            jar = YoutubeDLCookieJar()
182 183 184 185 186
            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}')
P
pukkandan 已提交
187
                    cookie = http.cookiejar.Cookie(
188 189 190 191 192
                        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)
P
pukkandan 已提交
193
            logger.info(f'Extracted {len(jar)} cookies from firefox')
194 195 196 197 198 199 200
            return jar
        finally:
            if cursor is not None:
                cursor.connection.close()


def _firefox_browser_dir():
201
    if sys.platform in ('cygwin', 'win32'):
202
        return os.path.expandvars(R'%APPDATA%\Mozilla\Firefox\Profiles')
203 204
    elif sys.platform == 'darwin':
        return os.path.expanduser('~/Library/Application Support/Firefox')
205
    return os.path.expanduser('~/.mozilla/firefox')
206 207 208 209


def _get_chromium_based_browser_settings(browser_name):
    # https://chromium.googlesource.com/chromium/src/+/HEAD/docs/user_data_dir.md
210
    if sys.platform in ('cygwin', 'win32'):
211 212 213
        appdata_local = os.path.expandvars('%LOCALAPPDATA%')
        appdata_roaming = os.path.expandvars('%APPDATA%')
        browser_dir = {
214 215 216 217 218 219
            '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'),
220 221 222 223 224 225 226 227 228 229 230 231 232 233
        }[browser_name]

    elif sys.platform == 'darwin':
        appdata = os.path.expanduser('~/Library/Application Support')
        browser_dir = {
            'brave': os.path.join(appdata, 'BraveSoftware/Brave-Browser'),
            'chrome': os.path.join(appdata, 'Google/Chrome'),
            'chromium': os.path.join(appdata, 'Chromium'),
            'edge': os.path.join(appdata, 'Microsoft Edge'),
            'opera': os.path.join(appdata, 'com.operasoftware.Opera'),
            'vivaldi': os.path.join(appdata, 'Vivaldi'),
        }[browser_name]

    else:
234 235 236 237 238 239 240 241 242
        config = _config_home()
        browser_dir = {
            'brave': os.path.join(config, 'BraveSoftware/Brave-Browser'),
            'chrome': os.path.join(config, 'google-chrome'),
            'chromium': os.path.join(config, 'chromium'),
            'edge': os.path.join(config, 'microsoft-edge'),
            'opera': os.path.join(config, 'opera'),
            'vivaldi': os.path.join(config, 'vivaldi'),
        }[browser_name]
243 244 245 246 247 248 249

    # 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"
    keyring_name = {
        'brave': 'Brave',
        'chrome': 'Chrome',
        'chromium': 'Chromium',
P
pukkandan 已提交
250
        'edge': 'Microsoft Edge' if sys.platform == 'darwin' else 'Chromium',
251 252 253 254 255 256 257 258 259 260 261 262 263
        'opera': 'Opera' if sys.platform == 'darwin' else 'Chromium',
        'vivaldi': 'Vivaldi' if sys.platform == 'darwin' else 'Chrome',
    }[browser_name]

    browsers_without_profiles = {'opera'}

    return {
        'browser_dir': browser_dir,
        'keyring_name': keyring_name,
        'supports_profiles': browser_name not in browsers_without_profiles
    }


264
def _extract_chrome_cookies(browser_name, profile, keyring, logger):
P
pukkandan 已提交
265
    logger.info(f'Extracting cookies from {browser_name}')
266

267
    if not sqlite3:
268 269
        logger.warning(f'Cannot extract cookies from {browser_name} without sqlite3 support. '
                       'Please use a python interpreter compiled with sqlite3 support')
270 271
        return YoutubeDLCookieJar()

272 273 274 275 276 277 278 279 280 281 282
    config = _get_chromium_based_browser_settings(browser_name)

    if profile is None:
        search_root = config['browser_dir']
    elif _is_path(profile):
        search_root = profile
        config['browser_dir'] = os.path.dirname(profile) if config['supports_profiles'] else profile
    else:
        if config['supports_profiles']:
            search_root = os.path.join(config['browser_dir'], profile)
        else:
P
pukkandan 已提交
283
            logger.error(f'{browser_name} does not support profiles')
284 285
            search_root = config['browser_dir']

286
    cookie_database_path = _find_most_recently_used_file(search_root, 'Cookies', logger)
287
    if cookie_database_path is None:
P
pukkandan 已提交
288 289
        raise FileNotFoundError(f'could not find {browser_name} cookies database in "{search_root}"')
    logger.debug(f'Extracting cookies from: "{cookie_database_path}"')
290

291
    decryptor = get_cookie_decryptor(config['browser_dir'], config['keyring_name'], logger, keyring=keyring)
292

293
    with tempfile.TemporaryDirectory(prefix='yt_dlp') as tmpdir:
294 295 296 297 298 299
        cursor = None
        try:
            cursor = _open_database_copy(cookie_database_path, tmpdir)
            cursor.connection.text_factory = bytes
            column_names = _get_column_names(cursor, 'cookies')
            secure_column = 'is_secure' if 'is_secure' in column_names else 'secure'
300
            cursor.execute(f'SELECT host_key, name, value, encrypted_value, path, expires_utc, {secure_column} FROM cookies')
301 302
            jar = YoutubeDLCookieJar()
            failed_cookies = 0
303
            unencrypted_cookies = 0
304 305 306 307 308 309 310
            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:
311 312
                        failed_cookies += 1
                        continue
313 314 315
                    elif not is_encrypted:
                        unencrypted_cookies += 1
                    jar.set_cookie(cookie)
316
            if failed_cookies > 0:
P
pukkandan 已提交
317
                failed_message = f' ({failed_cookies} could not be decrypted)'
318 319
            else:
                failed_message = ''
P
pukkandan 已提交
320
            logger.info(f'Extracted {len(jar)} cookies from {browser_name}{failed_message}')
P
pukkandan 已提交
321
            counts = decryptor._cookie_counts.copy()
322
            counts['unencrypted'] = unencrypted_cookies
P
pukkandan 已提交
323
            logger.debug(f'cookie version breakdown: {counts}')
324 325 326 327 328 329
            return jar
        finally:
            if cursor is not None:
                cursor.connection.close()


330
def _process_chrome_cookie(decryptor, host_key, name, value, encrypted_value, path, expires_utc, is_secure):
P
pukkandan 已提交
331 332 333 334
    host_key = host_key.decode()
    name = name.decode()
    value = value.decode()
    path = path.decode()
335 336 337 338 339 340 341
    is_encrypted = not value and encrypted_value

    if is_encrypted:
        value = decryptor.decrypt(encrypted_value)
        if value is None:
            return is_encrypted, None

P
pukkandan 已提交
342
    return is_encrypted, http.cookiejar.Cookie(
343 344 345 346 347 348
        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={})


349 350 351 352 353 354 355
class ChromeCookieDecryptor:
    """
    Overview:

        Linux:
        - cookies are either v10 or v11
            - v10: AES-CBC encrypted with a fixed key
356
                - also attempts empty password if decryption fails
357
            - v11: AES-CBC encrypted with an OS protected key (keyring)
358
                - also attempts empty password if decryption fails
359 360 361 362 363 364 365 366 367 368 369 370 371 372
            - v11 keys can be stored in various places depending on the activate desktop environment [2]

        Mac:
        - cookies are either v10 or not v10
            - v10: AES-CBC encrypted with an OS protected key (keyring) and more key derivation iterations than linux
            - not v10: 'old data' stored as plaintext

        Windows:
        - cookies are either v10 or not v10
            - v10: AES-GCM encrypted with a key which is encrypted with DPAPI
            - not v10: encrypted with DPAPI

    Sources:
    - [1] https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/
373
    - [2] https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/key_storage_linux.cc
374 375 376
        - KeyStorageLinux::CreateService
    """

377
    _cookie_counts = {}
378

379
    def decrypt(self, encrypted_value):
380
        raise NotImplementedError('Must be implemented by sub classes')
381

382

383
def get_cookie_decryptor(browser_root, browser_keyring_name, logger, *, keyring=None):
P
pukkandan 已提交
384
    if sys.platform == 'darwin':
385
        return MacChromeCookieDecryptor(browser_keyring_name, logger)
P
pukkandan 已提交
386
    elif sys.platform in ('win32', 'cygwin'):
387
        return WindowsChromeCookieDecryptor(browser_root, logger)
P
pukkandan 已提交
388
    return LinuxChromeCookieDecryptor(browser_keyring_name, logger, keyring=keyring)
389 390 391


class LinuxChromeCookieDecryptor(ChromeCookieDecryptor):
392
    def __init__(self, browser_keyring_name, logger, *, keyring=None):
393 394
        self._logger = logger
        self._v10_key = self.derive_key(b'peanuts')
395
        self._empty_key = self.derive_key(b'')
396
        self._cookie_counts = {'v10': 0, 'v11': 0, 'other': 0}
397 398 399 400 401 402 403
        self._browser_keyring_name = browser_keyring_name
        self._keyring = keyring

    @functools.cached_property
    def _v11_key(self):
        password = _get_linux_keyring_password(self._browser_keyring_name, self._keyring, self._logger)
        return None if password is None else self.derive_key(password)
404 405 406 407

    @staticmethod
    def derive_key(password):
        # values from
408
        # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/os_crypt_linux.cc
409 410 411
        return pbkdf2_sha1(password, salt=b'saltysalt', iterations=1, key_length=16)

    def decrypt(self, encrypted_value):
412 413 414 415 416 417 418 419 420 421
        """

        following the same approach as the fix in [1]: if cookies fail to decrypt then attempt to decrypt
        with an empty password. The failure detection is not the same as what chromium uses so the
        results won't be perfect

        References:
            - [1] https://chromium.googlesource.com/chromium/src/+/bbd54702284caca1f92d656fdcadf2ccca6f4165%5E%21/
                - a bugfix to try an empty password as a fallback
        """
422 423 424 425
        version = encrypted_value[:3]
        ciphertext = encrypted_value[3:]

        if version == b'v10':
426
            self._cookie_counts['v10'] += 1
427
            return _decrypt_aes_cbc_multi(ciphertext, (self._v10_key, self._empty_key), self._logger)
428 429

        elif version == b'v11':
430
            self._cookie_counts['v11'] += 1
431
            if self._v11_key is None:
432
                self._logger.warning('cannot decrypt v11 cookies: no key found', only_once=True)
433
                return None
434
            return _decrypt_aes_cbc_multi(ciphertext, (self._v11_key, self._empty_key), self._logger)
435 436

        else:
437
            self._logger.warning(f'unknown cookie version: "{version}"', only_once=True)
438
            self._cookie_counts['other'] += 1
439 440 441 442 443 444
            return None


class MacChromeCookieDecryptor(ChromeCookieDecryptor):
    def __init__(self, browser_keyring_name, logger):
        self._logger = logger
445
        password = _get_mac_keyring_password(browser_keyring_name, logger)
446
        self._v10_key = None if password is None else self.derive_key(password)
447
        self._cookie_counts = {'v10': 0, 'other': 0}
448 449 450 451

    @staticmethod
    def derive_key(password):
        # values from
452
        # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/os_crypt_mac.mm
453 454 455 456 457 458 459
        return pbkdf2_sha1(password, salt=b'saltysalt', iterations=1003, key_length=16)

    def decrypt(self, encrypted_value):
        version = encrypted_value[:3]
        ciphertext = encrypted_value[3:]

        if version == b'v10':
460
            self._cookie_counts['v10'] += 1
461 462 463 464
            if self._v10_key is None:
                self._logger.warning('cannot decrypt v10 cookies: no key found', only_once=True)
                return None

465
            return _decrypt_aes_cbc_multi(ciphertext, (self._v10_key,), self._logger)
466 467

        else:
468
            self._cookie_counts['other'] += 1
469
            # other prefixes are considered 'old data' which were stored as plaintext
470
            # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/os_crypt_mac.mm
471 472 473 474 475 476 477
            return encrypted_value


class WindowsChromeCookieDecryptor(ChromeCookieDecryptor):
    def __init__(self, browser_root, logger):
        self._logger = logger
        self._v10_key = _get_windows_v10_key(browser_root, logger)
478 479
        self._cookie_counts = {'v10': 0, 'other': 0}

480 481 482 483 484
    def decrypt(self, encrypted_value):
        version = encrypted_value[:3]
        ciphertext = encrypted_value[3:]

        if version == b'v10':
485
            self._cookie_counts['v10'] += 1
486 487 488 489
            if self._v10_key is None:
                self._logger.warning('cannot decrypt v10 cookies: no key found', only_once=True)
                return None

490
            # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/os_crypt_win.cc
491 492 493 494 495 496 497 498 499 500 501 502 503 504
            #   kNonceLength
            nonce_length = 96 // 8
            # boringssl
            #   EVP_AEAD_AES_GCM_TAG_LEN
            authentication_tag_length = 16

            raw_ciphertext = ciphertext
            nonce = raw_ciphertext[:nonce_length]
            ciphertext = raw_ciphertext[nonce_length:-authentication_tag_length]
            authentication_tag = raw_ciphertext[-authentication_tag_length:]

            return _decrypt_aes_gcm(ciphertext, self._v10_key, nonce, authentication_tag, self._logger)

        else:
505
            self._cookie_counts['other'] += 1
506
            # any other prefix means the data is DPAPI encrypted
507
            # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/os_crypt_win.cc
P
pukkandan 已提交
508
            return _decrypt_windows_dpapi(encrypted_value, self._logger).decode()
509 510 511 512


def _extract_safari_cookies(profile, logger):
    if sys.platform != 'darwin':
P
pukkandan 已提交
513
        raise ValueError(f'unsupported platform: {sys.platform}')
514

515 516 517 518 519 520 521
    if profile:
        cookies_path = os.path.expanduser(profile)
        if not os.path.isfile(cookies_path):
            raise FileNotFoundError('custom safari cookies database not found')

    else:
        cookies_path = os.path.expanduser('~/Library/Cookies/Cookies.binarycookies')
522

523
        if not os.path.isfile(cookies_path):
524 525 526 527
            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')
528 529 530 531 532

    with open(cookies_path, 'rb') as f:
        cookies_data = f.read()

    jar = parse_safari_cookies(cookies_data, logger=logger)
P
pukkandan 已提交
533
    logger.info(f'Extracted {len(jar)} cookies from safari')
534 535 536 537 538 539 540 541 542 543 544 545 546 547 548
    return jar


class ParserError(Exception):
    pass


class DataParser:
    def __init__(self, data, logger):
        self._data = data
        self.cursor = 0
        self._logger = logger

    def read_bytes(self, num_bytes):
        if num_bytes < 0:
P
pukkandan 已提交
549
            raise ParserError(f'invalid read of {num_bytes} bytes')
550 551 552 553 554 555 556 557 558 559
        end = self.cursor + num_bytes
        if end > len(self._data):
            raise ParserError('reached end of input')
        data = self._data[self.cursor:end]
        self.cursor = end
        return data

    def expect_bytes(self, expected_value, message):
        value = self.read_bytes(len(expected_value))
        if value != expected_value:
P
pukkandan 已提交
560
            raise ParserError(f'unexpected value: {value} != {expected_value} ({message})')
561 562 563 564 565 566 567 568 569 570 571 572 573 574

    def read_uint(self, big_endian=False):
        data_format = '>I' if big_endian else '<I'
        return struct.unpack(data_format, self.read_bytes(4))[0]

    def read_double(self, big_endian=False):
        data_format = '>d' if big_endian else '<d'
        return struct.unpack(data_format, self.read_bytes(8))[0]

    def read_cstring(self):
        buffer = []
        while True:
            c = self.read_bytes(1)
            if c == b'\x00':
P
pukkandan 已提交
575
                return b''.join(buffer).decode()
576 577 578 579 580
            else:
                buffer.append(c)

    def skip(self, num_bytes, description='unknown'):
        if num_bytes > 0:
581
            self._logger.debug(f'skipping {num_bytes} bytes ({description}): {self.read_bytes(num_bytes)!r}')
582
        elif num_bytes < 0:
P
pukkandan 已提交
583
            raise ParserError(f'invalid skip of {num_bytes} bytes')
584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609

    def skip_to(self, offset, description='unknown'):
        self.skip(offset - self.cursor, description)

    def skip_to_end(self, description='unknown'):
        self.skip_to(len(self._data), description)


def _mac_absolute_time_to_posix(timestamp):
    return int((datetime(2001, 1, 1, 0, 0, tzinfo=timezone.utc) + timedelta(seconds=timestamp)).timestamp())


def _parse_safari_cookies_header(data, logger):
    p = DataParser(data, logger)
    p.expect_bytes(b'cook', 'database signature')
    number_of_pages = p.read_uint(big_endian=True)
    page_sizes = [p.read_uint(big_endian=True) for _ in range(number_of_pages)]
    return page_sizes, p.cursor


def _parse_safari_cookies_page(data, jar, logger):
    p = DataParser(data, logger)
    p.expect_bytes(b'\x00\x00\x01\x00', 'page signature')
    number_of_cookies = p.read_uint()
    record_offsets = [p.read_uint() for _ in range(number_of_cookies)]
    if number_of_cookies == 0:
P
pukkandan 已提交
610
        logger.debug(f'a cookies page of size {len(data)} has no cookies')
611 612 613 614
        return

    p.skip_to(record_offsets[0], 'unknown page header field')

615 616 617 618 619 620
    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)
621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651
    p.skip_to_end('space in between pages')


def _parse_safari_cookies_record(data, jar, logger):
    p = DataParser(data, logger)
    record_size = p.read_uint()
    p.skip(4, 'unknown record field 1')
    flags = p.read_uint()
    is_secure = bool(flags & 0x0001)
    p.skip(4, 'unknown record field 2')
    domain_offset = p.read_uint()
    name_offset = p.read_uint()
    path_offset = p.read_uint()
    value_offset = p.read_uint()
    p.skip(8, 'unknown record field 3')
    expiration_date = _mac_absolute_time_to_posix(p.read_double())
    _creation_date = _mac_absolute_time_to_posix(p.read_double())  # noqa: F841

    try:
        p.skip_to(domain_offset)
        domain = p.read_cstring()

        p.skip_to(name_offset)
        name = p.read_cstring()

        p.skip_to(path_offset)
        path = p.read_cstring()

        p.skip_to(value_offset)
        value = p.read_cstring()
    except UnicodeDecodeError:
652
        logger.warning('failed to parse Safari cookie because UTF-8 decoding failed', only_once=True)
653 654 655 656
        return record_size

    p.skip_to(record_size, 'space at the end of the record')

P
pukkandan 已提交
657
    cookie = http.cookiejar.Cookie(
658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682
        version=0, name=name, value=value, port=None, port_specified=False,
        domain=domain, domain_specified=bool(domain), domain_initial_dot=domain.startswith('.'),
        path=path, path_specified=bool(path), secure=is_secure, expires=expiration_date, discard=False,
        comment=None, comment_url=None, rest={})
    jar.set_cookie(cookie)
    return record_size


def parse_safari_cookies(data, jar=None, logger=YDLLogger()):
    """
    References:
        - https://github.com/libyal/dtformats/blob/main/documentation/Safari%20Cookies.asciidoc
            - this data appears to be out of date but the important parts of the database structure is the same
            - there are a few bytes here and there which are skipped during parsing
    """
    if jar is None:
        jar = YoutubeDLCookieJar()
    page_sizes, body_start = _parse_safari_cookies_header(data, logger)
    p = DataParser(data[body_start:], logger)
    for page_size in page_sizes:
        _parse_safari_cookies_page(p.read_bytes(page_size), jar, logger)
    p.skip_to_end('footer')
    return jar


683 684 685 686 687 688 689
class _LinuxDesktopEnvironment(Enum):
    """
    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/base/nix/xdg_util.h
    DesktopEnvironment
    """
    OTHER = auto()
    CINNAMON = auto()
690
    DEEPIN = auto()
691
    GNOME = auto()
692 693 694 695
    KDE3 = auto()
    KDE4 = auto()
    KDE5 = auto()
    KDE6 = auto()
696
    PANTHEON = auto()
697
    UKUI = auto()
698 699
    UNITY = auto()
    XFCE = auto()
700
    LXQT = auto()
701 702


703 704
class _LinuxKeyring(Enum):
    """
705
    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/key_storage_util_linux.h
706 707
    SelectedLinuxBackend
    """
708 709 710 711 712
    KWALLET4 = auto()  # this value is just called KWALLET in the chromium source but it is for KDE4 only
    KWALLET5 = auto()
    KWALLET6 = auto()
    GNOME_KEYRING = auto()
    BASIC_TEXT = auto()
713 714 715 716 717


SUPPORTED_KEYRINGS = _LinuxKeyring.__members__.keys()


718
def _get_linux_desktop_environment(env, logger):
719 720 721 722 723 724 725 726 727 728 729 730 731 732
    """
    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
733 734
        elif xdg_current_desktop == 'Deepin':
            return _LinuxDesktopEnvironment.DEEPIN
735 736 737 738 739
        elif xdg_current_desktop == 'GNOME':
            return _LinuxDesktopEnvironment.GNOME
        elif xdg_current_desktop == 'X-Cinnamon':
            return _LinuxDesktopEnvironment.CINNAMON
        elif xdg_current_desktop == 'KDE':
740 741 742 743 744 745 746 747 748 749
            kde_version = env.get('KDE_SESSION_VERSION', None)
            if kde_version == '5':
                return _LinuxDesktopEnvironment.KDE5
            elif kde_version == '6':
                return _LinuxDesktopEnvironment.KDE6
            elif kde_version == '4':
                return _LinuxDesktopEnvironment.KDE4
            else:
                logger.info(f'unknown KDE version: "{kde_version}". Assuming KDE4')
                return _LinuxDesktopEnvironment.KDE4
750 751 752 753
        elif xdg_current_desktop == 'Pantheon':
            return _LinuxDesktopEnvironment.PANTHEON
        elif xdg_current_desktop == 'XFCE':
            return _LinuxDesktopEnvironment.XFCE
754 755 756 757 758 759 760
        elif xdg_current_desktop == 'UKUI':
            return _LinuxDesktopEnvironment.UKUI
        elif xdg_current_desktop == 'LXQt':
            return _LinuxDesktopEnvironment.LXQT
        else:
            logger.info(f'XDG_CURRENT_DESKTOP is set to an unknown value: "{xdg_current_desktop}"')

761
    elif desktop_session is not None:
762 763 764
        if desktop_session == 'deepin':
            return _LinuxDesktopEnvironment.DEEPIN
        elif desktop_session in ('mate', 'gnome'):
765
            return _LinuxDesktopEnvironment.GNOME
766 767 768 769 770 771 772 773
        elif desktop_session in ('kde4', 'kde-plasma'):
            return _LinuxDesktopEnvironment.KDE4
        elif desktop_session == 'kde':
            if 'KDE_SESSION_VERSION' in env:
                return _LinuxDesktopEnvironment.KDE4
            else:
                return _LinuxDesktopEnvironment.KDE3
        elif 'xfce' in desktop_session or desktop_session == 'xubuntu':
774
            return _LinuxDesktopEnvironment.XFCE
775 776 777 778 779
        elif desktop_session == 'ukui':
            return _LinuxDesktopEnvironment.UKUI
        else:
            logger.info(f'DESKTOP_SESSION is set to an unknown value: "{desktop_session}"')

780 781 782 783
    else:
        if 'GNOME_DESKTOP_SESSION_ID' in env:
            return _LinuxDesktopEnvironment.GNOME
        elif 'KDE_FULL_SESSION' in env:
784 785 786 787
            if 'KDE_SESSION_VERSION' in env:
                return _LinuxDesktopEnvironment.KDE4
            else:
                return _LinuxDesktopEnvironment.KDE3
788
    return _LinuxDesktopEnvironment.OTHER
789 790 791 792


def _choose_linux_keyring(logger):
    """
793 794 795 796 797 798 799 800 801
    SelectBackend in [1]

    There is currently support for forcing chromium to use BASIC_TEXT by creating a file called
    `Disable Local Encryption` [1] in the user data dir. The function to write this file (`WriteBackendUse()` [1])
    does not appear to be called anywhere other than in tests, so the user would have to create this file manually
    and so would be aware enough to tell yt-dlp to use the BASIC_TEXT keyring.

    References:
        - [1] https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/key_storage_util_linux.cc
802
    """
803
    desktop_environment = _get_linux_desktop_environment(os.environ, logger)
P
pukkandan 已提交
804
    logger.debug(f'detected desktop environment: {desktop_environment.name}')
805 806 807 808 809 810 811 812 813 814
    if desktop_environment == _LinuxDesktopEnvironment.KDE4:
        linux_keyring = _LinuxKeyring.KWALLET4
    elif desktop_environment == _LinuxDesktopEnvironment.KDE5:
        linux_keyring = _LinuxKeyring.KWALLET5
    elif desktop_environment == _LinuxDesktopEnvironment.KDE6:
        linux_keyring = _LinuxKeyring.KWALLET6
    elif desktop_environment in (
        _LinuxDesktopEnvironment.KDE3, _LinuxDesktopEnvironment.LXQT, _LinuxDesktopEnvironment.OTHER
    ):
        linux_keyring = _LinuxKeyring.BASIC_TEXT
815
    else:
816
        linux_keyring = _LinuxKeyring.GNOME_KEYRING
817 818 819
    return linux_keyring


820
def _get_kwallet_network_wallet(keyring, logger):
821 822
    """ The name of the wallet used to store network passwords.

823
    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/kwallet_dbus.cc
824 825 826 827 828 829 830
    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:
831 832 833 834 835 836 837 838 839 840 841 842
        if keyring == _LinuxKeyring.KWALLET4:
            service_name = 'org.kde.kwalletd'
            wallet_path = '/modules/kwalletd'
        elif keyring == _LinuxKeyring.KWALLET5:
            service_name = 'org.kde.kwalletd5'
            wallet_path = '/modules/kwalletd5'
        elif keyring == _LinuxKeyring.KWALLET6:
            service_name = 'org.kde.kwalletd6'
            wallet_path = '/modules/kwalletd6'
        else:
            raise ValueError(keyring)

843
        stdout, _, returncode = Popen.run([
844
            'dbus-send', '--session', '--print-reply=literal',
845 846
            f'--dest={service_name}',
            wallet_path,
847
            'org.kde.KWallet.networkWallet'
848
        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
849

850
        if returncode:
851 852 853
            logger.warning('failed to read NetworkWallet')
            return default_wallet
        else:
854 855
            logger.debug(f'NetworkWallet = "{stdout.strip()}"')
            return stdout.strip()
P
pukkandan 已提交
856
    except Exception as e:
P
pukkandan 已提交
857
        logger.warning(f'exception while obtaining NetworkWallet: {e}')
858 859 860
        return default_wallet


861 862
def _get_kwallet_password(browser_keyring_name, keyring, logger):
    logger.debug(f'using kwallet-query to obtain password from {keyring.name}')
863 864 865 866 867 868 869

    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''

870
    network_wallet = _get_kwallet_network_wallet(keyring, logger)
871 872

    try:
873
        stdout, _, returncode = Popen.run([
874
            'kwallet-query',
P
pukkandan 已提交
875 876
            '--read-password', f'{browser_keyring_name} Safe Storage',
            '--folder', f'{browser_keyring_name} Keys',
877 878 879
            network_wallet
        ], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)

880 881 882
        if returncode:
            logger.error(f'kwallet-query failed with return code {returncode}. '
                         'Please consult the kwallet-query man page for details')
883 884 885 886 887 888 889 890 891
            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.
892 893 894
                # this was identified as a bug later and fixed in
                # https://chromium.googlesource.com/chromium/src/+/bbd54702284caca1f92d656fdcadf2ccca6f4165%5E%21/#F0
                # https://chromium.googlesource.com/chromium/src/+/5463af3c39d7f5b6d11db7fbd51e38cc1974d764
895 896 897
                return b''
            else:
                logger.debug('password found')
898
                return stdout.rstrip(b'\n')
P
pukkandan 已提交
899 900
    except Exception as e:
        logger.warning(f'exception running kwallet-query: {error_to_str(e)}')
901 902 903 904
        return b''


def _get_gnome_keyring_password(browser_keyring_name, logger):
905 906
    if not secretstorage:
        logger.error(f'secretstorage not available {_SECRETSTORAGE_UNAVAILABLE_REASON}')
907 908 909 910 911 912 913 914
        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():
P
pukkandan 已提交
915
            if item.get_label() == f'{browser_keyring_name} Safe Storage':
916 917 918 919 920 921 922 923 924 925 926 927 928
                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.

929
    keyring = _LinuxKeyring[keyring] if keyring else _choose_linux_keyring(logger)
930 931
    logger.debug(f'Chosen keyring: {keyring.name}')

932 933 934
    if keyring in (_LinuxKeyring.KWALLET4, _LinuxKeyring.KWALLET5, _LinuxKeyring.KWALLET6):
        return _get_kwallet_password(browser_keyring_name, keyring, logger)
    elif keyring == _LinuxKeyring.GNOME_KEYRING:
935
        return _get_gnome_keyring_password(browser_keyring_name, logger)
936
    elif keyring == _LinuxKeyring.BASIC_TEXT:
937 938 939 940 941 942 943 944
        # 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:
945
        stdout, _, returncode = Popen.run(
946 947 948
            ['security', 'find-generic-password',
             '-w',  # write password to stdout
             '-a', browser_keyring_name,  # match 'account'
P
pukkandan 已提交
949
             '-s', f'{browser_keyring_name} Safe Storage'],  # match 'service'
950
            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
951 952 953
        if returncode:
            logger.warning('find-generic-password failed')
            return None
954
        return stdout.rstrip(b'\n')
P
pukkandan 已提交
955 956
    except Exception as e:
        logger.warning(f'exception running find-generic-password: {error_to_str(e)}')
957
        return None
958 959 960


def _get_windows_v10_key(browser_root, logger):
961 962 963 964
    """
    References:
        - [1] https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/sync/os_crypt_win.cc
    """
965
    path = _find_most_recently_used_file(browser_root, 'Local State', logger)
966 967 968
    if path is None:
        logger.error('could not find local state file')
        return None
969
    logger.debug(f'Found local state file at "{path}"')
P
pukkandan 已提交
970
    with open(path, encoding='utf8') as f:
971 972
        data = json.load(f)
    try:
973
        # kOsCryptEncryptedKeyPrefName in [1]
974 975 976 977
        base64_key = data['os_crypt']['encrypted_key']
    except KeyError:
        logger.error('no encrypted key in Local State')
        return None
978
    encrypted_key = base64.b64decode(base64_key)
979
    # kDPAPIKeyPrefix in [1]
980 981 982 983 984 985 986 987 988 989 990
    prefix = b'DPAPI'
    if not encrypted_key.startswith(prefix):
        logger.error('invalid key')
        return None
    return _decrypt_windows_dpapi(encrypted_key[len(prefix):], logger)


def pbkdf2_sha1(password, salt, iterations, key_length):
    return pbkdf2_hmac('sha1', password, salt, iterations, key_length)


991 992 993 994 995 996 997 998 999
def _decrypt_aes_cbc_multi(ciphertext, keys, logger, initialization_vector=b' ' * 16):
    for key in keys:
        plaintext = unpad_pkcs7(aes_cbc_decrypt_bytes(ciphertext, key, initialization_vector))
        try:
            return plaintext.decode()
        except UnicodeDecodeError:
            pass
    logger.warning('failed to decrypt cookie (AES-CBC) because UTF-8 decoding failed. Possibly the key is wrong?', only_once=True)
    return None
1000 1001 1002 1003


def _decrypt_aes_gcm(ciphertext, key, nonce, authentication_tag, logger):
    try:
1004
        plaintext = aes_gcm_decrypt_and_verify_bytes(ciphertext, key, authentication_tag, nonce)
1005
    except ValueError:
1006
        logger.warning('failed to decrypt cookie (AES-GCM) because the MAC check failed. Possibly the key is wrong?', only_once=True)
1007 1008 1009
        return None

    try:
P
pukkandan 已提交
1010
        return plaintext.decode()
1011
    except UnicodeDecodeError:
1012
        logger.warning('failed to decrypt cookie (AES-GCM) because UTF-8 decoding failed. Possibly the key is wrong?', only_once=True)
1013 1014 1015 1016 1017 1018 1019 1020
        return None


def _decrypt_windows_dpapi(ciphertext, logger):
    """
    References:
        - https://docs.microsoft.com/en-us/windows/win32/api/dpapi/nf-dpapi-cryptunprotectdata
    """
P
pukkandan 已提交
1021 1022 1023

    import ctypes
    import ctypes.wintypes
1024 1025

    class DATA_BLOB(ctypes.Structure):
P
pukkandan 已提交
1026
        _fields_ = [('cbData', ctypes.wintypes.DWORD),
1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041
                    ('pbData', ctypes.POINTER(ctypes.c_char))]

    buffer = ctypes.create_string_buffer(ciphertext)
    blob_in = DATA_BLOB(ctypes.sizeof(buffer), buffer)
    blob_out = DATA_BLOB()
    ret = ctypes.windll.crypt32.CryptUnprotectData(
        ctypes.byref(blob_in),  # pDataIn
        None,  # ppszDataDescr: human readable description of pDataIn
        None,  # pOptionalEntropy: salt?
        None,  # pvReserved: must be NULL
        None,  # pPromptStruct: information about prompts to display
        0,  # dwFlags
        ctypes.byref(blob_out)  # pDataOut
    )
    if not ret:
1042
        logger.warning('failed to decrypt with DPAPI', only_once=True)
1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062
        return None

    result = ctypes.string_at(blob_out.pbData, blob_out.cbData)
    ctypes.windll.kernel32.LocalFree(blob_out.pbData)
    return result


def _config_home():
    return os.environ.get('XDG_CONFIG_HOME', os.path.expanduser('~/.config'))


def _open_database_copy(database_path, tmpdir):
    # cannot open sqlite databases if they are already in use (e.g. by the browser)
    database_copy_path = os.path.join(tmpdir, 'temporary.sqlite')
    shutil.copy(database_path, database_copy_path)
    conn = sqlite3.connect(database_copy_path)
    return conn.cursor()


def _get_column_names(cursor, table_name):
P
pukkandan 已提交
1063
    table_info = cursor.execute(f'PRAGMA table_info({table_name})').fetchall()
P
pukkandan 已提交
1064
    return [row[1].decode() for row in table_info]
1065 1066


1067
def _find_most_recently_used_file(root, filename, logger):
1068
    # if there are multiple browser profiles, take the most recently used one
1069 1070 1071 1072 1073 1074 1075 1076
    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))
1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093
    return None if not paths else max(paths, key=lambda path: os.lstat(path).st_mtime)


def _merge_cookie_jars(jars):
    output_jar = YoutubeDLCookieJar()
    for jar in jars:
        for cookie in jar:
            output_jar.set_cookie(cookie)
        if jar.filename is not None:
            output_jar.filename = jar.filename
    return output_jar


def _is_path(value):
    return os.path.sep in value


1094
def _parse_browser_specification(browser_name, profile=None, keyring=None, container=None):
1095 1096
    if browser_name not in SUPPORTED_BROWSERS:
        raise ValueError(f'unsupported browser: "{browser_name}"')
1097 1098
    if keyring not in (None, *SUPPORTED_KEYRINGS):
        raise ValueError(f'unsupported keyring: "{keyring}"')
1099 1100
    if profile is not None and _is_path(expand_path(profile)):
        profile = expand_path(profile)
1101
    return browser_name, profile, keyring, container
1102 1103 1104 1105 1106


class LenientSimpleCookie(http.cookies.SimpleCookie):
    """More lenient version of http.cookies.SimpleCookie"""
    # From https://github.com/python/cpython/blob/v3.10.7/Lib/http/cookies.py
1107 1108 1109
    # We use Morsel's legal key chars to avoid errors on setting values
    _LEGAL_KEY_CHARS = r'\w\d' + re.escape('!#$%&\'*+-.:^_`|~')
    _LEGAL_VALUE_CHARS = _LEGAL_KEY_CHARS + re.escape('(),/<=>?@[]{}')
1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154

    _RESERVED = {
        "expires",
        "path",
        "comment",
        "domain",
        "max-age",
        "secure",
        "httponly",
        "version",
        "samesite",
    }

    _FLAGS = {"secure", "httponly"}

    # Added 'bad' group to catch the remaining value
    _COOKIE_PATTERN = re.compile(r"""
        \s*                            # Optional whitespace at start of cookie
        (?P<key>                       # Start of group 'key'
        [""" + _LEGAL_KEY_CHARS + r"""]+?# Any word of at least one letter
        )                              # End of group 'key'
        (                              # Optional group: there may not be a value.
        \s*=\s*                          # Equal Sign
        (                                # Start of potential value
        (?P<val>                           # Start of group 'val'
        "(?:[^\\"]|\\.)*"                    # Any doublequoted string
        |                                    # or
        \w{3},\s[\w\d\s-]{9,11}\s[\d:]{8}\sGMT # Special case for "expires" attr
        |                                    # or
        [""" + _LEGAL_VALUE_CHARS + r"""]*     # Any word or empty string
        )                                  # End of group 'val'
        |                                  # or
        (?P<bad>(?:\\;|[^;])*?)            # 'bad' group fallback for invalid values
        )                                # End of potential value
        )?                             # End of optional value group
        \s*                            # Any number of spaces.
        (\s+|;|$)                      # Ending either at space, semicolon, or EOS.
        """, re.ASCII | re.VERBOSE)

    def load(self, data):
        # Workaround for https://github.com/yt-dlp/yt-dlp/issues/4776
        if not isinstance(data, str):
            return super().load(data)

        morsel = None
1155 1156
        for match in self._COOKIE_PATTERN.finditer(data):
            if match.group('bad'):
1157 1158 1159
                morsel = None
                continue

1160
            key, value = match.group('key', 'val')
1161

1162 1163 1164 1165
            is_attribute = False
            if key.startswith('$'):
                key = key[1:]
                is_attribute = True
1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181

            lower_key = key.lower()
            if lower_key in self._RESERVED:
                if morsel is None:
                    continue

                if value is None:
                    if lower_key not in self._FLAGS:
                        morsel = None
                        continue
                    value = True
                else:
                    value, _ = self.value_decode(value)

                morsel[key] = value

1182 1183 1184
            elif is_attribute:
                morsel = None

1185 1186 1187 1188 1189 1190 1191 1192
            elif value is not None:
                morsel = self.get(key, http.cookies.Morsel())
                real_value, coded_value = self.value_decode(value)
                morsel.set(key, real_value, coded_value)
                self[key] = morsel

            else:
                morsel = None
1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328


class YoutubeDLCookieJar(http.cookiejar.MozillaCookieJar):
    """
    See [1] for cookie file format.

    1. https://curl.haxx.se/docs/http-cookies.html
    """
    _HTTPONLY_PREFIX = '#HttpOnly_'
    _ENTRY_LEN = 7
    _HEADER = '''# Netscape HTTP Cookie File
# This file is generated by yt-dlp.  Do not edit.

'''
    _CookieFileEntry = collections.namedtuple(
        'CookieFileEntry',
        ('domain_name', 'include_subdomains', 'path', 'https_only', 'expires_at', 'name', 'value'))

    def __init__(self, filename=None, *args, **kwargs):
        super().__init__(None, *args, **kwargs)
        if is_path_like(filename):
            filename = os.fspath(filename)
        self.filename = filename

    @staticmethod
    def _true_or_false(cndn):
        return 'TRUE' if cndn else 'FALSE'

    @contextlib.contextmanager
    def open(self, file, *, write=False):
        if is_path_like(file):
            with open(file, 'w' if write else 'r', encoding='utf-8') as f:
                yield f
        else:
            if write:
                file.truncate(0)
            yield file

    def _really_save(self, f, ignore_discard=False, ignore_expires=False):
        now = time.time()
        for cookie in self:
            if (not ignore_discard and cookie.discard
                    or not ignore_expires and cookie.is_expired(now)):
                continue
            name, value = cookie.name, cookie.value
            if value is None:
                # cookies.txt regards 'Set-Cookie: foo' as a cookie
                # with no name, whereas http.cookiejar regards it as a
                # cookie with no value.
                name, value = '', name
            f.write('%s\n' % '\t'.join((
                cookie.domain,
                self._true_or_false(cookie.domain.startswith('.')),
                cookie.path,
                self._true_or_false(cookie.secure),
                str_or_none(cookie.expires, default=''),
                name, value
            )))

    def save(self, filename=None, *args, **kwargs):
        """
        Save cookies to a file.
        Code is taken from CPython 3.6
        https://github.com/python/cpython/blob/8d999cbf4adea053be6dbb612b9844635c4dfb8e/Lib/http/cookiejar.py#L2091-L2117 """

        if filename is None:
            if self.filename is not None:
                filename = self.filename
            else:
                raise ValueError(http.cookiejar.MISSING_FILENAME_TEXT)

        # Store session cookies with `expires` set to 0 instead of an empty string
        for cookie in self:
            if cookie.expires is None:
                cookie.expires = 0

        with self.open(filename, write=True) as f:
            f.write(self._HEADER)
            self._really_save(f, *args, **kwargs)

    def load(self, filename=None, ignore_discard=False, ignore_expires=False):
        """Load cookies from a file."""
        if filename is None:
            if self.filename is not None:
                filename = self.filename
            else:
                raise ValueError(http.cookiejar.MISSING_FILENAME_TEXT)

        def prepare_line(line):
            if line.startswith(self._HTTPONLY_PREFIX):
                line = line[len(self._HTTPONLY_PREFIX):]
            # comments and empty lines are fine
            if line.startswith('#') or not line.strip():
                return line
            cookie_list = line.split('\t')
            if len(cookie_list) != self._ENTRY_LEN:
                raise http.cookiejar.LoadError('invalid length %d' % len(cookie_list))
            cookie = self._CookieFileEntry(*cookie_list)
            if cookie.expires_at and not cookie.expires_at.isdigit():
                raise http.cookiejar.LoadError('invalid expires at %s' % cookie.expires_at)
            return line

        cf = io.StringIO()
        with self.open(filename) as f:
            for line in f:
                try:
                    cf.write(prepare_line(line))
                except http.cookiejar.LoadError as e:
                    if f'{line.strip()} '[0] in '[{"':
                        raise http.cookiejar.LoadError(
                            'Cookies file must be Netscape formatted, not JSON. See  '
                            'https://github.com/yt-dlp/yt-dlp/wiki/FAQ#how-do-i-pass-cookies-to-yt-dlp')
                    write_string(f'WARNING: skipping cookie file entry due to {e}: {line!r}\n')
                    continue
        cf.seek(0)
        self._really_load(cf, filename, ignore_discard, ignore_expires)
        # Session cookies are denoted by either `expires` field set to
        # an empty string or 0. MozillaCookieJar only recognizes the former
        # (see [1]). So we need force the latter to be recognized as session
        # cookies on our own.
        # Session cookies may be important for cookies-based authentication,
        # e.g. usually, when user does not check 'Remember me' check box while
        # logging in on a site, some important cookies are stored as session
        # cookies so that not recognizing them will result in failed login.
        # 1. https://bugs.python.org/issue17164
        for cookie in self:
            # Treat `expires=0` cookies as session cookies
            if cookie.expires == 0:
                cookie.expires = None
                cookie.discard = True

    def get_cookie_header(self, url):
        """Generate a Cookie HTTP header for a given url"""
        cookie_req = urllib.request.Request(escape_url(sanitize_url(url)))
        self.add_cookie_header(cookie_req)
        return cookie_req.get_header('Cookie')