webdriver.py

# Licensed to the Software Freedom Conservancy (SFC) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The SFC licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

# modified by kaliiiiiiiiii | Aurin Aegerter
# all modifications are licensed under the license provided at LICENSE.md

"""The WebDriver implementation."""
import os
import shutil
import subprocess
import sys
import tempfile
import time
import typing
import uuid
import warnings
import signal
from typing import List

# io
import asyncio

import cdp_socket.exceptions
import websockets

# interactions
from selenium_driverless.input.pointer import Pointer
from selenium_driverless.types.webelement import WebElement
from selenium_driverless.scripts.switch_to import SwitchTo

# contexts
from selenium_driverless.sync.context import Context as SyncContext
from selenium_driverless.types.context import Context

# Targets
from selenium_driverless.scripts.driver_utils import get_target
from selenium_driverless.types.target import Target, TargetInfo
from selenium_driverless.types.base_target import BaseTarget
from selenium_driverless.sync.base_target import BaseTarget as SyncBaseTarget

# others
from cdp_socket.utils.conn import get_json
from selenium_driverless.types.options import Options as ChromeOptions
from selenium_driverless.utils.utils import sel_driverless_path
from selenium_driverless.types import JSEvalException
from selenium_driverless import EXC_HANDLER


class Chrome:
    """Control the chromium based browsers without any driver."""
    port: int

    def __init__(
            self,
            options: ChromeOptions = None,
            timeout: float = 30,
            debug: bool = False,
            max_ws_size: int = 2 ** 27
    ) -> None:
        # noinspection GrazieInspection
        """Creates a new instance of the chrome target. Starts the service and
                then creates new instance of chrome target.

                .. code-block:: python

                    options = webdriver.ChromeOptions.rst()
                    async with webdriver.Chrome(options=options) as driver:
                        await driver.get('https://abrahamjuliot.github.io/creepjs/', wait_load=True)
                        print(await driver.title)

                :param options: this takes an instance of ChromeOptions.rst
                :param timeout: timeout in seconds to start chrome
                :param debug: redirect errors from the chromium process output (stderr) to console
                :param max_ws_size: maximum size for websocket messages in bytes. 2^27 ~= 130 MB by default
                """
        self._prefs = {}
        self._auth_interception_enabled = None
        self._mv3_extension = None
        self._extensions_incognito_allowed = None
        self._base_context = None
        self._stderr = None
        self._stderr_file = None
        self._process = None
        self._current_target = None
        self._host = None
        self._timeout = timeout
        self._loop: asyncio.AbstractEventLoop or None = None
        self.browser_pid: int or None = None
        self._base_target = None
        self._debug = debug
        # noinspection PyTypeChecker
        self._current_context: Context = None
        self._contexts: typing.Dict[str, Context] = {}
        self._temp_dir = tempfile.TemporaryDirectory(prefix="selenium_driverless_").name
        self._max_ws_size = max_ws_size

        self._auth = {}

        if not options:
            options = ChromeOptions()
        if not options.binary_location:
            from selenium_driverless.utils.utils import find_chrome_executable
            options.binary_location = find_chrome_executable()

        self._options: ChromeOptions = options
        self._is_remote = True
        self._is_remote = False
        self._has_incognito_contexts: bool = False
        self._started = False

    def __repr__(self):
        return f'<{type(self).__module__}.{type(self).__name__} (session="{self.current_target.id}")>'

    async def __aenter__(self):
        await self.start_session()
        return self

    def __enter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.quit(clean_dirs=self._options.auto_clean_dirs)

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def __await__(self):
        return self.start_session().__await__()

    async def start_session(self):
        if not self._started:
            from selenium_driverless.utils.utils import read
            from selenium_driverless.utils.utils import is_first_run, get_default_ua, set_default_ua
            from selenium_driverless.scripts.prefs import read_prefs, write_prefs

            await is_first_run()
            user_agent = await get_default_ua()

            if self._options.use_extension:
                # extension
                self._options.add_extension(sel_driverless_path() + "files/mv3_extension")

            if not self._options.debugger_address:
                from selenium_driverless.utils.utils import random_port
                port = random_port()
                self._options._debugger_address = f"127.0.0.1:{port}"
                self._options.add_argument(f"--remote-debugging-port={port}")

            if self._options.headless and not self._is_remote:
                # patch useragent
                if user_agent:
                    self._options.add_argument(f"--user-agent={user_agent}")
                else:
                    warnings.warn("headless is detectable at first run")

            # handle prefs

            if self._options.user_data_dir:
                prefs_path = self._options.user_data_dir + "/Default/Preferences"
                if os.path.isfile(prefs_path):
                    self._prefs = await read_prefs(prefs_path)
                else:
                    os.makedirs(os.path.dirname(prefs_path), exist_ok=True)

                # write prefs
                self._prefs.update(self._options.prefs)
                await write_prefs(self._prefs, prefs_path)
            elif self._options.user_data_dir is None:
                self._options.add_argument(
                    "--user-data-dir=" + self._temp_dir + "/data_dir")
                prefs_path = self._options.user_data_dir + "/Default/Preferences"
                os.makedirs(os.path.dirname(prefs_path), exist_ok=True)

                # write prefs
                self._prefs.update(self._options.prefs)
                await write_prefs(self._prefs, prefs_path)

            # noinspection PyProtectedMember
            # handle extensions
            if self._options._extension_paths:
                import zipfile
                extension_paths = []
                loop = asyncio.get_running_loop()

                # noinspection PyProtectedMember
                def extractall():
                    for _path in self._options._extension_paths:
                        if os.path.isfile(_path):
                            with zipfile.ZipFile(_path, 'r') as zip_ref:
                                _path = self._temp_dir + f"/{uuid.uuid4().hex}"
                                zip_ref.extractall(_path)
                        extension_paths.append(_path)

                await loop.run_in_executor(None, extractall)

                self._options.arguments.append(f"--load-extension=" + ','.join(extension_paths))
            self._options._extension_paths = []

            if self._options.startup_url:
                self._options.add_argument(self._options.startup_url)
            self._options._startup_url = None

            options = self._options

            # noinspection PyProtectedMember
            self._is_remote = self._options._is_remote

            if not self._is_remote:
                path = options.binary_location
                args = options.arguments
                if self._debug:
                    self._stderr = sys.stderr
                else:
                    self._stderr = tempfile.TemporaryFile(prefix="selenium_driverless")
                    self._stderr_file = self._stderr
                self._process = subprocess.Popen(
                    [path, *args],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=self._stderr,
                    close_fds=True,
                    preexec_fn=os.setsid if os.name == 'posix' else None,
                    creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == 'nt' else 0,
                    shell=False,
                    text=True,
                    env=self._options.env
                )

                host, port = self._options.debugger_address.split(":")
                port = int(port)
                if port == 0:
                    path = self._options.user_data_dir + "/DevToolsActivePort"
                    while not os.path.isfile(path):
                        await asyncio.sleep(0.1)
                    port = await read(path, sel_root=False)
                    port = int(port.split("\n")[0])
                    self._options.debugger_address = f"127.0.0.1:{port}"

            host, port = self._options.debugger_address.split(":")
            self.port = int(port)
            self._host = f"{host}:{self.port}"
            if self._loop:
                self._base_target = await SyncBaseTarget(host=self._host, is_remote=self._is_remote,
                                                         timeout=self._timeout, loop=self._loop,
                                                         max_ws_size=self._max_ws_size)
            else:
                self._base_target = await BaseTarget(host=self._host, is_remote=self._is_remote,
                                                     timeout=self._timeout, loop=self._loop,
                                                     max_ws_size=self._max_ws_size)

            # fetch useragent at first headless run
            # noinspection PyUnboundLocalVariable
            if not self._is_remote:
                res = await self._base_target.execute_cdp_cmd("Browser.getVersion")
                user_agent = res["userAgent"]
                user_agent = user_agent.replace("HeadlessChrome", "Chrome")
                await set_default_ua(user_agent)

            if not self._is_remote:
                # noinspection PyUnboundLocalVariable
                self.browser_pid = self._process.pid
            targets = await get_json(self._host, timeout=self._timeout)
            for target in targets:
                if target["type"] == "page" and not target["url"].startswith("chrome-extension://"):
                    target_id = target["id"]
                    self._current_target = await get_target(target_id=target_id, host=self._host,
                                                            loop=self._loop, is_remote=self._is_remote, timeout=10,
                                                            max_ws_size=self._max_ws_size, driver=self, context=None)

                    # handle the context
                    if self._loop:
                        context = await SyncContext(base_target=self._current_target, driver=self, loop=self._loop,
                                                    max_ws_size=self._max_ws_size)
                    else:
                        context = await Context(base_target=self._current_target, driver=self, loop=self._loop,
                                                max_ws_size=self._max_ws_size)
                    _id = context.context_id
                    self._current_target._context = context

                    def remove_context():
                        if _id in self._contexts:
                            del self._contexts[_id]
                        self._base_context = None

                    # noinspection PyProtectedMember
                    context._closed_callbacks.append(remove_context)
                    self._current_context = context
                    self._base_context = context
                    self._contexts[_id] = context
                    break
            await self.execute_cdp_cmd("Emulation.setFocusEmulationEnabled", {"enabled": True})
            if self._options.single_proxy:
                await self.set_single_proxy(self._options.single_proxy)
            downloads_dir = self._options.downloads_dir
            if self._options.downloads_dir:
                # ensure download events are dispatched
                await self.set_download_behaviour("allowAndName", downloads_dir)
            else:
                await self.set_download_behaviour("default")
            self._started = True
        return self

    @property
    async def frame_tree(self) -> dict:
        """
        **async**
        all nested frames within the current target
        """
        return await self.current_context.frame_tree

    @property
    async def targets(self) -> typing.Dict[str, TargetInfo]:
        """
        **async**
        all targets within the current context
        """
        return await self.current_context.targets

    @property
    async def contexts(self) -> typing.Dict[str, Context]:
        """
        **async**
        all (incognito) contexts on Chrome.
        """
        targets = await self.get_targets(context_id=None)
        contexts = {}
        for info in targets.values():
            _id = info.browser_context_id
            if _id:
                context = self._contexts.get(_id)
                if not context:
                    if self._loop:
                        context = await SyncContext(base_target=self._current_target, context_id=_id,
                                                    loop=self._loop, max_ws_size=self._max_ws_size, driver=self)
                    else:
                        context = await Context(base_target=self._current_target, context_id=_id,
                                                loop=self._loop, max_ws_size=self._max_ws_size, driver=self)
                contexts[_id] = context
        self._contexts.update(contexts)
        return self._contexts

    async def new_context(self, proxy_bypass_list: typing.List[str] = None, proxy_server: str = True,
                          universal_access_origins: typing.List[str] = None, url: str = "about:blank") -> Context:
        """
        creates a new (incognito) context

        :param url: the url the first tab will start at. "about:blank" by default
        :param universal_access_origins: An optional list of origins to grant unlimited cross-origin access to. Parts of the URL other than those constituting origin are ignored

        .. warning::
            The proxy parameter doesn't work on Windows due to `crbug#1310057 <https://bugs.chromium.org/p/chromium/issues/detail?id=1310057>`__.

        .. code block:: python
            await driver.set_auth("username", "password", "localhost:5000")
            context = await driver.new_context(proxy_bypass_list=["localhost"], proxy_server="http://localhost:5000")

        :param proxy_server: a proxy-server to use for the context
        :param proxy_bypass_list: a list of proxies to ignore
        """
        await self.ensure_extensions_incognito_allowed()
        if proxy_bypass_list is None:
            proxy_bypass_list = ["localhost"]
        if proxy_server is None:
            proxy_server = ""
        args = {"disposeOnDetach": False}
        if proxy_bypass_list:
            args["proxyBypassList"] = ",".join(proxy_bypass_list)
        if not (proxy_server is True):
            args["proxyServer"] = proxy_server
        if universal_access_origins:
            args["originsWithUniversalNetworkAccess"] = universal_access_origins

        # create context ensuring extension racing conditions
        self._auth_interception_enabled = False
        has_incognito_ctxs = not (not self._has_incognito_contexts)
        self._has_incognito_contexts = True
        mv3_ext = await self.mv3_extension
        self._mv3_extension = None

        try:
            res = await self.base_target.execute_cdp_cmd("Target.createBrowserContext", args)
        except Exception as e:
            self._mv3_extension = mv3_ext
            self._auth_interception_enabled = True
            self._has_incognito_contexts = has_incognito_ctxs
            raise e

        _id = res["browserContextId"]
        if self._loop:
            context = await SyncContext(base_target=self._base_target, context_id=_id, loop=self._loop,
                                        is_incognito=True,
                                        max_ws_size=self._max_ws_size, driver=self)
        else:
            context = await Context(base_target=self._base_target, context_id=_id, loop=self._loop,
                                    is_incognito=True, max_ws_size=self._max_ws_size,
                                    driver=self)
        self._contexts[_id] = context

        def remove_context():
            if _id in self._contexts:
                del self._contexts[_id]

        # noinspection PyProtectedMember
        context._closed_callbacks.append(remove_context)
        await context.switch_to.new_window("window", activate=False, url=url)
        tabs = await context.get_targets(_type="page", context_id=_id)
        context._current_target = list(tabs.values())[0].Target
        context._current_target._timeout = 5

        # reload auth & extension target to fix non-applied auth
        if self._auth:
            self._mv3_extension = None
            while True:
                # ensure racing conditions with extension
                try:
                    mv3_target = await self.mv3_extension
                    self._auth_interception_enabled = False
                    await self._ensure_auth_interception(timeout=0.5, set_flag=False)
                    await mv3_target.execute_script("globalThis.authCreds = arguments[0]", self._auth, timeout=0.5,
                                                    unique_context=False)
                except (asyncio.TimeoutError, TimeoutError):
                    await asyncio.sleep(0.1)
                    self._mv3_extension = None
                else:
                    self._auth_interception_enabled = True
                    self._mv3_extension = mv3_target
                    return context
        return context

    async def get_targets(self,
                          _type: typing.Literal["page", "background_page", "service_worker", "browser", "other"] = None,
                          context_id: str or None = "self") -> typing.Dict[str, TargetInfo]:
        """
        get all targets within the current context
        :param _type: filter by target type
        :param context_id: if ``None``, this function returns all targets for all contexts.
        """
        return await self.current_context.get_targets(_type=_type, context_id=context_id)

    @property
    def current_target(self) -> Target:
        """
        the current Target
        """
        if self.current_context:
            return self.current_context.current_target
        return self._current_target

    @property
    def base_target(self) -> BaseTarget:
        """
        The connection handle for the global connection to Chrome

        .. warning::
            only the bindings for using the CDP-protocol on BaseTarget supported
        """
        return self._base_target

    @property
    async def mv3_extension(self, timeout: float = 10) -> Target:
        """
        **async** the target for the background script of the by default loaded Chrome-extension (manifest-version==3)

        .. note:
            for incognito context, the extension uses the "spanning" configuration, as there isn't a way to debug "split" mode over CDP
        """
        if self._has_incognito_contexts:
            await self.ensure_extensions_incognito_allowed()
        if not self._mv3_extension:
            import re
            import time
            start = time.perf_counter()
            extension_target = None
            while not extension_target:
                targets = await self.get_targets(context_id=None)
                for target in targets.values():
                    if target.type == "service_worker":
                        if re.fullmatch(
                                r"chrome-extension://(.*)/"
                                r"driverless_background_mv3_243ffdd55e32a012b4f253b2879af978\.js",
                                target.url):
                            extension_target = target.Target
                            break
                if not extension_target:
                    if (time.perf_counter() - start) > timeout:
                        raise asyncio.TimeoutError(f"Couldn't find mv3 extension within {timeout} seconds")
            while True:
                try:
                    # fix WebRTC leak
                    await extension_target.execute_script(
                        "chrome.privacy.network.webRTCIPHandlingPolicy.set(arguments[0])",
                        {"value": "disable_non_proxied_udp"}, timeout=2, unique_context=False)
                except (asyncio.TimeoutError, TimeoutError):
                    await asyncio.sleep(0.2)
                    return await self.mv3_extension
                except JSEvalException:
                    await asyncio.sleep(0.2)
                except cdp_socket.exceptions.CDPError as e:
                    if e.code == -32000 and e.message == 'Could not find object with given id':
                        await asyncio.sleep(0.2)
                else:
                    break
            self._mv3_extension = extension_target
        return self._mv3_extension

    async def ensure_extensions_incognito_allowed(self):
        """
        ensure that all installed Chrome-extensions are allowed in incognito context.

        .. warning::
            Generally, the extension decides whether to use the ``split``, ``spanning`` or ``not_allowed`` configuration.
            For changing this behaviour, you'll have to modify the ``manifest.json`` file within the compressed extension or directory.
            See `developer.chrome.com/docs/extensions/reference/manifest/incognito <https://developer.chrome.com/docs/extensions/reference/manifest/incognito?hl=en>`__.
        """
        if not self._extensions_incognito_allowed:
            self._extensions_incognito_allowed = True
            # noinspection PyTypeChecker
            page = None
            try:
                base_ctx = self._base_context
                page: Context = await base_ctx.new_window("tab", "chrome://extensions", activate=False)
                script = """
                    async function make_global(){
                        const extensions = await chrome.developerPrivate.getExtensionsInfo();
                        extensions.forEach( async function(extension)  {
                            chrome.developerPrivate.updateExtensionConfiguration({
                                extensionId: extension.id,
                                incognitoAccess: true
                            })
                        });
                    };
                    await make_global()
                """
                await asyncio.sleep(0.1)
                await page.eval_async(script, timeout=10, unique_context=False)
            except Exception as e:
                EXC_HANDLER(e)
                self._extensions_incognito_allowed = False
                if page:
                    await page.close()
                await self.ensure_extensions_incognito_allowed()
            self._extensions_incognito_allowed = True
            await page.close()

    @property
    def base_context(self) -> Context:
        """
        the Context which isn't incognito
        """
        return self._base_context

    @property
    def downloads_dir(self):
        """the current downloads directory for the current context"""
        return self.base_target.downloads_dir_for_context(context_id="DEFAULT")

    async def set_download_behaviour(self, behaviour: typing.Literal["deny", "allow", "allowAndName", "default"],
                                     path: str = None):
        """set the download behaviour

        :param behaviour: the behaviour to set the downloading to
        :param path: the path to the default download directory

        .. warning::
            setting ``behaviour=allow`` instead of ``allowAndName`` can cause some bugs

        """
        await self.current_context.set_download_behaviour(behaviour, path)

    @property
    def current_context(self) -> Context:
        """
        the current context switched to
        """
        if not self._current_context:
            if self._contexts:
                return list(self._contexts.values())[0]
        return self._current_context

    @property
    async def _isolated_context_id(self):
        # noinspection PyProtectedMember
        return await self.current_context._isolated_context_id

    async def get_target(self, target_id: str = None, timeout: float = 2) -> Target:
        """
        get a Target by TargetId for advanced usage of the CDP protocol
        :param target_id:
        :param timeout: timeout in seconds for connecting to the target if it's not tracked already
        """
        if not target_id:
            return self.current_target
        return await self.current_context.get_target(target_id=target_id, timeout=timeout)

    async def get_target_for_iframe(self, iframe: WebElement) -> Target:
        """
        get the Target for a specific iframe

        .. warning::
            only cross-iframes have a Target due to `OOPIF <https://www.chromium.org/developers/design-documents/oop-iframes/>`__. See `site-isolation <https://www.chromium.org/Home/chromium-security/site-isolation/>`__
            For a general solution, have a look at ``WebElement.content_document`` instead

        :param iframe: the iframe to get the Target for
        """
        return await self.current_target.get_target_for_iframe(iframe=iframe)

    async def get_targets_for_iframes(self, iframes: typing.List[WebElement]) -> typing.List[Target]:
        """
        returns a list of targets for iframes
        see ``webdriver.Chrome.get_target_for_iframe`` for more information

        :param iframes: the iframe to get the targets for
        """
        return await self.current_target.get_targets_for_iframes(iframes=iframes)

    async def wait_download(self, timeout: float or None = 30) -> dict:
        """
        wait for a download on the current tab

        returns something like

        .. code-block:: python

            {
                "frameId": "2D543B5E8B14945B280C537A4882A695",
                "guid": "c91df4d5-9b45-4962-84df-3749bd3f926d",
                "url": "https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf",
                "suggestedFilename": "dummy.pdf",

                # only if options.downloads_dir specified
                "guid_file": "D:\\System\\AppData\\PyCharm\\scratches\\downloads\\c91df4d5-9b45-4962-84df-3749bd3f926d"
            }

        :param timeout: time in seconds to wait for a download

        .. warning::
            downloads from iframes not supported yet

        """
        return await self.current_target.wait_download(timeout=timeout)

    async def get(self, url: str, referrer: str = None, wait_load: bool = True, timeout: float = 30) -> typing.Union[
        None, dict]:
        """Loads a web page in the current Target

        :param url: the url to load.
        :param referrer: the referrer to load the page with
        :param wait_load: whether to wait for the webpage to load
        :param timeout: the maximum time in seconds for waiting on load

        returns the same as :func:`Target.wait_download <selenium_driverless.types.target.Target.wait_download>` if the url initiates a download
        """
        return await self.current_target.get(url=url, referrer=referrer, wait_load=wait_load, timeout=timeout)

    @property
    async def title(self) -> str:
        """**async** the title of the current target"""
        target = await self.current_target_info
        return target.title

    @property
    def current_pointer(self) -> Pointer:
        """the :class:`Pointer <selenium_driverless.input.pointer.Pointer>` for this target"""
        target = self.current_target
        return target.pointer

    async def send_keys(self, text: str):
        """
        send text & keys to the current target

        :param text: the text to send
        """
        await self.current_target.send_keys(text)

    async def execute_raw_script(self, script: str, *args, await_res: bool = False,
                                 serialization: typing.Literal["deep", "json", "idOnly"] = "deep",
                                 max_depth: int = None, timeout: float = 2, execution_context_id,
                                 unique_context: bool = True):
        """executes a JavaScript on ``GlobalThis`` such as

        .. code-block:: js

            function(...arguments){return document}

        :param script: the script as a string
        :param args: the argument which are passed to the function. Those can be either json-serializable or a RemoteObject such as WebElement
        :param await_res: whether to await the function or the return value of it
        :param serialization: can be one of ``deep``, ``json``, ``idOnly``
        :param max_depth: The maximum depth objects get serialized.
        :param timeout: the maximum time to wait for the execution to complete
        :param execution_context_id: the execution context id to run the JavaScript in. Exclusive with unique_context
        :param unique_context: whether to use an isolated context to run the Script in.

        see `Runtime.callFunctionOn <https://chromedevtools.github.io/devtools-protocol/tot/Runtime/#method-callFunctionOn>`_
        """
        return await self.current_target.execute_raw_script(script, *args, await_res=await_res,
                                                            serialization=serialization, max_depth=max_depth,
                                                            timeout=timeout, execution_context_id=execution_context_id,
                                                            unique_context=unique_context)

    async def execute_script(self, script: str, *args, max_depth: int = 2, serialization: str = None,
                             timeout: float = 2, execution_context_id: str = None,
                             unique_context: bool = True):
        """executes JavaScript synchronously on ``GlobalThis`` such as

        .. code-block:: js

            return document

        see :func:`Target.execute_raw_script <selenium_driverless.types.target.Target.execute_raw_script>` for argument descriptions
                """
        return await self.current_target.execute_script(script, *args, max_depth=max_depth, serialization=serialization,
                                                        timeout=timeout, execution_context_id=execution_context_id,
                                                        unique_context=unique_context)

    async def execute_async_script(self, script: str, *args, max_depth: int = 2,
                                   serialization: str = None, timeout: float = 2, execution_context_id: str = None,
                                   unique_context: bool = True):
        """executes JavaScript asynchronously on ``GlobalThis`` such as

        .. warning::
            using execute_async_script is not recommended as it doesn't handle exceptions correctly.
            Use :func:`Chrome.eval_async <selenium_driverless.webdriver.Chrome.eval_async>`

        .. code-block:: js

            resolve = arguments[arguments.length-1]

        see :func:`Target.execute_raw_script <selenium_driverless.types.target.Target.execute_raw_script>` for argument descriptions
        """
        return await self.current_target.execute_async_script(script, *args, max_depth=max_depth,
                                                              serialization=serialization,
                                                              timeout=timeout,
                                                              execution_context_id=execution_context_id,
                                                              unique_context=unique_context)

    async def eval_async(self, script: str, *args, max_depth: int = 2,
                         serialization: str = None, timeout: float = 2, execution_context_id: str = None,
                         unique_context: bool = True):
        """executes JavaScript asynchronously on ``GlobalThis`` such as

        .. code-block:: js

            res = await fetch("https://httpbin.org/get");
            // mind CORS!
            json = await res.json()
            return json

        see :func:`Target.execute_raw_script <selenium_driverless.types.target.Target.execute_raw_script>` for argument descriptions
        """
        return await self.current_target.eval_async(script, *args, max_depth=max_depth, serialization=serialization,
                                                    timeout=timeout,
                                                    execution_context_id=execution_context_id,
                                                    unique_context=unique_context)

    @property
    async def current_url(self) -> str:
        """**async** current URL of the current Target
        """
        target = self.current_target
        return await target.url

    @property
    async def page_source(self) -> str:
        """**async** html of the current page.
        """
        target = self.current_target
        return await target.page_source

    async def close(self, timeout: float = 2) -> None:
        """Closes the current target (only works for tabs).
        :param timeout: timeout in seconds for the tab to close
        """
        await self.current_target.close(timeout=timeout)

    async def focus(self):
        """focuses the current target (only works for tabs)
        """
        await self.current_target.focus()

    async def quit(self, timeout: float = 30, clean_dirs: bool = True) -> None:
        """Closes Chrome
        :param timeout: the maximum time waiting for chrome to quit correctly
        :param clean_dirs: whether to clean out the user-data-dir directory
        """
        from selenium_driverless import EXC_HANDLER

        loop = asyncio.get_running_loop()

        def clean_dirs_sync(dirs: typing.List[str]):
            for _dir in dirs:
                while os.path.isdir(_dir):
                    shutil.rmtree(_dir, ignore_errors=True)

        if self._started:
            start = time.perf_counter()
            # noinspection PyUnresolvedReferences
            try:
                # assumption: chrome is still running
                await self.base_target.execute_cdp_cmd("Browser.close", timeout=7)
            except websockets.ConnectionClosedError:
                pass
            except Exception as e:
                EXC_HANDLER(e)
            if not self._is_remote:
                if self._process is not None:
                    # assumption: chrome is being shutdown manually or programmatically
                    try:
                        await loop.run_in_executor(None, lambda: self._process.wait(timeout))
                    except Exception as e:
                        EXC_HANDLER(e)
                    else:
                        self._process = None
                try:
                    # assumption: chrome hasn't closed within timeout, killing with force
                    # wait for process to be killed
                    if self._process is not None:
                        if os.name == 'posix':
                            os.killpg(os.getpgid(self._process.pid), signal.SIGTERM)
                        else:
                            self._process.terminate()
                        try:
                            await loop.run_in_executor(None, lambda: self._process.wait(timeout))
                        except subprocess.TimeoutExpired:
                            if os.name == 'posix':
                                os.killpg(os.getpgid(self._process.pid), signal.SIGKILL)
                            else:
                                self._process.kill()
                except Exception as e:
                    EXC_HANDLER(e)
                finally:
                    self._started = False
                    if self._stderr_file:
                        try:
                            self._stderr.close()
                        except Exception as e:
                            EXC_HANDLER(e)

                    # clean temp dir for extensions etc
                    try:
                        await asyncio.wait_for(
                            # wait for
                            loop.run_in_executor(None,
                                                 lambda: clean_dirs_sync(
                                                     [self._temp_dir])),
                            timeout=max(5, int(timeout - (time.perf_counter() - start))))
                    except Exception as e:
                        EXC_HANDLER(e)

                    if clean_dirs:
                        # clean user-data-dir for chrome
                        try:
                            await asyncio.wait_for(
                                # wait for
                                loop.run_in_executor(None,
                                                     lambda: clean_dirs_sync(
                                                         [self._options.user_data_dir])),
                                timeout=max(5, int(timeout - (time.perf_counter() - start))))
                        except Exception as e:
                            warnings.warn(
                                "driver hasn't quit correctly, "
                                "files might be left in your temp folder & chrome might still be running",
                                ResourceWarning)
                            raise e

    def __del__(self):
        try:
            if self._started:
                warnings.warn(
                    "driver hasn't quit correctly, "
                    "files might be left in your temp folder & chrome might still be running",
                    ResourceWarning)
        except AttributeError:
            pass

    @property
    async def current_target_info(self) -> TargetInfo:
        """**async** TargetInfo of the current target"""
        return await self.current_target.info

    @property
    def current_window_handle(self) -> str:
        """current TargetId

        .. warning::

            this is deprecated and will be removed
            use ``webdriver.Chrome.current_target.id`` instead
        """
        warnings.warn(f'"webdriver.Chrome.current_window_handle" is deprecated and will be removed\n'
                      'use "webdriver.Chrome.current_target.id" instead', DeprecationWarning)
        if self.current_target:
            return self.current_target.id

    @property
    async def current_window_id(self):
        """**async** the ``WindowId`` of the window the current Target belongs to
        """
        result = await self.execute_cdp_cmd("Browser.getWindowForTarget", {"targetId": self.current_target.id})
        return result["windowId"]

    @property
    async def window_handles(self) -> List[TargetInfo]:
        """**async** TargetInfo on all tabs in the current context

        .. warning::

            the tabs aren't ordered by position in the window. Do not rely on the index, but iterate and filter them.
        """
        warnings.warn(
            "the tabs aren't ordered by position in the window. Do not rely on the index, but iterate and filter them.")
        tabs = []
        targets = await self.targets
        for info in list(targets.values()):
            if info.type == "page":
                tabs.append(info)
        return tabs

    async def new_window(self, type_hint: typing.Literal["tab", "window"] = "tab", url="",
                         activate: bool = True) -> Target:
        """Creates a new window or tab in the current context

        :param type_hint: whether to create a tab or window
        :param url: the url which the new window should start on. Defaults to about:blank
        :param activate: whether to explicitly activate/focus the window

        .. code-block:: python

            new_target = driver.new_window('tab')
        """
        return await self.current_context.new_window(type_hint=type_hint, url=url, activate=activate)

    async def set_window_state(self, state: typing.Literal["normal", "minimized", "maximized", "fullscreen"]):
        """sets the window state on the window the current Target belongs to
        :param state: the state to set
        """
        window_id = await self.current_window_id
        bounds = {"windowState": state}
        await self.execute_cdp_cmd("Browser.setWindowBounds", {"bounds": bounds, "windowId": window_id})

    async def normalize_window(self):
        """Normalizes the window position and size on the window the current Target belongs to
        """
        await self.set_window_state("normal")

    async def maximize_window(self) -> None:
        """Maximizes the window the current Target belongs to"""
        await self.set_window_state("maximized")

    async def fullscreen_window(self) -> None:
        """enters fullscreen on the window the current Target belongs to"""
        await self.set_window_state("fullscreen")

    async def minimize_window(self) -> None:
        """minimizes the window the current Target belongs to.

        .. warning::

            Minimizing isn't recommended as it can throttle some functionalities in chrome.
        """
        await self.set_window_state("minimized")

    # noinspection PyUnusedLocal
    async def print_page(self) -> str:
        """Prints the page (current target => tab) to PDF
        """
        target = self.current_target
        return await target.print_page()

    @property
    def switch_to(self) -> SwitchTo:
        """SwitchTo handle
        """
        return self.current_context.switch_to

    # Navigation
    async def back(self) -> None:
        """Goes one step backward in the browser history on the current target (has to be a tab).
        """
        await self.current_target.back()

    async def forward(self) -> None:
        """Goes one step forward in the browser history on the current target (has to be a tab).
        """
        await self.current_target.forward()

    async def refresh(self) -> None:
        """Refreshes the current tab (target).
        """
        await self.current_target.refresh()

    # Options
    async def get_cookies(self) -> List[dict]:
        """list of cookies for the current tab
        """
        return await self.current_target.get_cookies()

    async def get_cookie(self, name) -> typing.Optional[typing.Dict]:
        """Get a single cookie by name. Returns the cookie if found, None if
        not.

        :param name: name of the cookie
        """
        return await self.current_target.get_cookie(name=name)

    async def delete_cookie(self, name: str, url: str = None, domain: str = None,
                            path: str = None) -> None:
        """Deletes a single cookie with the given name in the current tab.

        :param name: name of the cookie to delete
        :param url: url of the cookie
        :param domain: domain of the cookie
        :param path: path of the cookie
        """
        return await self.current_target.delete_cookie(name=name, url=url, domain=domain, path=path)

    async def delete_all_cookies(self) -> None:
        """Delete all cookies in the current (incognito-) context.
        """
        await self.current_target.delete_all_cookies()

    # noinspection GrazieInspection
    async def add_cookie(self, cookie_dict: dict) -> None:
        """Adds a cookie in the current (incognito-) context

        :param cookie_dict: see `Network.CookieParam <https://chromedevtools.github.io/devtools-protocol/tot/Network/#type-CookieParam>`__
        """
        await self.current_target.add_cookie(cookie_dict=cookie_dict)

    # Timeouts
    @staticmethod
    async def sleep(time_to_wait) -> None:
        # noinspection GrazieInspection
        """sleep
                .. note::
                    use this one instead of time.sleep in the sync version.

                :param time_to_wait: time in seconds to sleep
                """
        await asyncio.sleep(time_to_wait)

    # noinspection PyUnusedLocal
    async def find_element(self, by: str, value: str, timeout: float or None = None) -> WebElement:
        """find an element in the current target

        :param by: one of the locators at :func:`By <selenium_driverless.types.by.By>`
        :param value: the actual query to find the element by
        :param timeout: how long to wait for the element to exist
        """
        return await self.current_target.find_element(by=by, value=value, timeout=timeout)

    async def find_elements(self, by: str, value: str, timeout: float = 3) -> typing.List[WebElement]:
        """find multiple elements in the current target

        :param by: one of the locators at :func:`By <selenium_driverless.types.by.By>`
        :param value: the actual query to find the elements by
        :param timeout: how long to wait for not being in a page reload loop in seconds
        """
        return await self.current_target.find_elements(by=by, value=value, timeout=timeout)

    async def search_elements(self, query: str) -> typing.List[WebElement]:
        """
        find elements similarly to how "CTRL+F" in the DevTools Console works

        :param query: Plain text to find elements with
        """
        return await self.current_target.search_elements(query=query)

    async def get_screenshot_as_file(self, filename: str) -> None:
        """Saves a screenshot of the current tab to a PNG image file.
        :param filename: The path you wish to save your screenshot to. should end with a `.png` extension.

        .. code-block:: python

            driver.get_screenshot_as_file('screenshots/test.png')
        """
        return await self.current_target.get_screenshot_as_file(filename=filename)

    async def save_screenshot(self, filename) -> None:
        """alias to :func: `driver.get_screenshot_as_file <selenium_driverless.webdriver.Chrome.get_screenshot_as_file>`"""
        return await self.get_screenshot_as_file(filename)

    async def get_screenshot_as_png(self) -> bytes:
        """Gets the screenshot of the current tab as a binary data.
        """
        return await self.current_target.get_screenshot_as_png()

    async def snapshot(self) -> str:
        """gets the current snapshot as mhtml"""
        return await self.current_target.snapshot()

    async def save_snapshot(self, filename: str):
        """Saves a snapshot of the current window to a MHTML file.

        :param filename: The full path you wish to save your snapshot to. This
                   should end with a ``.mhtml`` extension.

        .. code-block:: Python

            await driver.get_snapshot('snapshot.mhtml')

        """
        return await self.current_target.save_snapshot(filename)

    # noinspection PyPep8Naming
    async def set_window_size(self, width: int, height: int) -> None:
        """Sets the width and height of the window, the current tab is within (unless ``windowHandle`` specified)

        :param width: the width in pixels to set the window to
        :param height: the height in pixels to set the window to
        """
        await self.set_window_rect(width=int(width), height=int(height))

    # noinspection PyPep8Naming
    async def get_window_size(self) -> dict:
        """Gets the width and height of the current window.

        returns something like:
        .. code-block: json

            {"width":1280, "height":720}
        """
        size = await self.get_window_rect()

        if size.get("value"):
            size = size["value"]

        return {k: size[k] for k in ("width", "height")}

    # noinspection PyPep8Naming
    async def set_window_position(self, x: int, y: int) -> dict:
        """Sets the x,y position of the window, the current tab is in.

        :param x: the x-coordinate in pixels to set the window position
        :param y: the y-coordinate in pixels to set the window position
        """
        return await self.set_window_rect(x=int(x), y=int(y))

    # noinspection PyPep8Naming
    async def get_window_position(self) -> dict:
        """Gets the x,y position of the window, the current tab is in.

        returns something like:
        .. code-block: json

            {"x":0, "y":0}
        """
        position = await self.get_window_rect()

        return {k: position[k] for k in ("x", "y")}

    async def get_window_rect(self) -> dict:
        """Gets the x, y, with and height coordinates of the window, the current tab is in.

        returns something like:
        .. code-block: json

            {"x":0, "y":0,
            "width":1280, "height":720,
            "windowState":"normal"
            }

        .. note::

            ``windowState`` can be one of "normal", "minimized", "maximized", "fullscreen"
        """
        json = await self.execute_cdp_cmd("Browser.getWindowBounds", {"windowId": await self.current_window_id})
        json = json["bounds"]
        json["x"] = json["left"]
        del json["left"]
        json["y"] = json["top"]
        del json["top"]
        return json

    async def set_window_rect(self, x=None, y=None, width=None, height=None) -> dict:
        """Sets the x, y, width and height coordinates of the window the current target is in.

        :param x: the x-coordinate in pixels to set the window position
        :param y: the y-coordinate in pixels to set the window position
        :param width: the width in pixels to set the window to
        :param height: the height in pixels to set the window to

        .. note::

            either x and y or with and height have to be specified
        """

        if (x is None and y is None) and (height is None and width is None):
            raise ValueError("x and y or height and width need values")

        bounds = {"left": x, "top": y, "width": width, 'height': height}
        await self.execute_cdp_cmd("Browser.setWindowBounds",
                                   {"windowId": await self.current_window_id, "bounds": bounds})
        bounds["x"] = bounds["left"]
        del bounds["left"]
        bounds["y"] = bounds["top"]
        del bounds["top"]

        return bounds

    async def get_network_conditions(self):
        """Gets Chromium network emulation settings.

        returns a dict like:

        .. code-block:: python

            {"latency": 4, "download_throughput": 2, "upload_throughput": 2,
            "offline": False}
        """
        return await self.current_target.get_network_conditions()

    # noinspection SpellCheckingInspection
    async def set_network_conditions(self, offline: bool, latency: int, download_throughput: int,
                                     upload_throughput: int,
                                     connection_type: typing.Literal[
                                         "none", "cellular2g", "cellular3g", "cellular4g", "bluetooth", "ethernet", "wifi", "wimax", "other"]) -> None:
        # noinspection GrazieInspection
        """Sets Chromium network emulation settings.

        :param offline:
        :param latency:  additional latency in ms
        :param download_throughput: maximum throughput, 500 * 1024 for example
        :param upload_throughput: maximum throughput, 500 * 1024 for example
        :param connection_type: the connection type

        .. note::

            'throughput' can be used to set both (for download and upload).
        """
        return await self.current_target.set_network_conditions(offline=offline, latency=latency,
                                                                download_throughput=download_throughput,
                                                                upload_throughput=upload_throughput,
                                                                connection_type=connection_type)

    async def delete_network_conditions(self) -> None:
        """Resets Chromium network emulation settings."""
        await self.current_target.delete_network_conditions()

    async def set_permissions(self, name: str, value: typing.Literal["granted", "denied", "prompt"],
                              origin: str = None) -> None:
        """Sets Applicable Permission

        :param name: The item to set the permission on.
        :param value: The value to set on the item
        :param origin: the origin the permission for. Applies to any origin if not set

        .. code-block:: python

            target.set_permissions('clipboard-read', 'denied')
        """
        settings = ["granted", "denied", "prompt"]
        if value not in settings:
            raise ValueError(f"value needs to be within {settings}, but got {value}")
        args = {"permission": {"name": name}, "setting": value}
        if origin:
            args["origin"] = origin
        await self.execute_cdp_cmd("Browser.setPermission", args)

    async def set_proxy(self, proxy_config):
        # noinspection GrazieInspection
        """ set a proxy dynamically

        Example parameters:

        .. code-block:: python

            proxy_config = {
                "mode": "fixed_servers",
                "rules": {
                    "proxyForHttp": {
                        "scheme": scheme,
                        "host": host,
                        "port": port
                    },
                    "proxyForHttps": {
                        "scheme": scheme,
                        "host": host,
                        "port": port
                    },
                    "proxyForFtp": {
                        "scheme": scheme,
                        "host": host,
                        "port": port
                    },
                    "fallbackProxy": {
                        "scheme": scheme,
                        "host": host,
                        "port": port
                    },
                    "bypassList": ["<local>"]
                }
            }

        :param proxy_config: see `developer.chrome.com/docs/extensions/reference/proxy <https://developer.chrome.com/docs/extensions/reference/proxy/>`__ for reference

        for authentification, see :func:`webdriver.Chrome.set_auth <selenium_driverless.webdriver.Chrome.set_auth>`
        """
        extension = await self.mv3_extension
        await extension.eval_async("await chrome.proxy.settings.set(arguments[0])",
                                   {"value": proxy_config, "scope": 'regular'}, unique_context=False)

    async def set_single_proxy(self, proxy: str, bypass_list=None):
        """
        Set a single proxy dynamically to be applied in all contexts.

        .. code-block:: python

            "http://user1:passwrd1@example.proxy.com:5001/"

        .. warning::

            - Only supported when Chrome has been started with driverless or the extension at ``selenium_driverless/files/mv3_extension`` has been loaded into the browser.

            - ``Socks5`` doesn't support authentication due to `crbug#1309413 <https://bugs.chromium.org/p/chromium/issues/detail?id=1309413>`__.

        """

        # parse scheme
        proxy = proxy.split("://")
        if len(proxy) == 2:
            scheme, proxy = proxy
        else:
            scheme = None
            proxy = proxy[0]

        proxy = proxy.split("@")
        if len(proxy) == 2:
            creds, proxy = proxy
        else:
            proxy = proxy[0]
            creds = None

        # parse host & port
        proxy = proxy.split(":")
        if len(proxy) == 2:
            host, port = proxy
            port = int(port.replace("/", ""))
        else:
            port = None
            host = proxy[0]

        rule = {"host": host}
        if scheme:
            rule["scheme"] = scheme
        if port:
            rule["port"] = port
        if bypass_list is None:
            bypass_list = ["<local>"]
        proxy_config = {
            "mode": "fixed_servers",
            "rules": {
                "proxyForHttp": rule,
                "proxyForHttps": rule,
                "proxyForFtp": rule,
                "fallbackProxy": rule,
                "bypassList": bypass_list
            }
        }
        await self.set_proxy(proxy_config)
        if creds:
            user, passw = creds.split(":")
            await self.set_auth(user, passw, f"{host}:{port}")

    async def clear_proxy(self):
        """
        Clear the applied proxy (=> use no proxy at all) in all contexts.
        """
        extension = await self.mv3_extension
        await extension.eval_async("""
            await chrome.proxy.settings.set(
              {value: {mode: "direct"}, scope: 'regular'}
            );
        """, unique_context=False)

    async def _ensure_auth_interception(self, timeout: float = 0.3, set_flag: bool = True):
        # internal, to re-apply auth interception which is broken when a new context gets opened. Due to how extensions in incognito work
        if not self._auth_interception_enabled:
            script = """
                        if(globalThis.authCreds == undefined){globalThis.authCreds = {}}
                        globalThis.onAuth = function onAuth(details) {
                            return globalThis.authCreds[details.challenger.host+":"+details.challenger.port]
                        }
                        chrome.webRequest.onAuthRequired.addListener(
                            onAuth,
                            {urls: ["<all_urls>"]},
                            ['blocking']
                            );
                        """
            mv3_target = await self.mv3_extension
            await mv3_target.execute_script(script, timeout=timeout, unique_context=False)
            if set_flag:
                self._auth_interception_enabled = True

    async def set_auth(self, username: str, password: str, host_with_port):
        """
        Set authentication dynamically to be applied in all contexts.

        .. code-block:: python

            driver.set_auth("user1","passwrd1", "example.com:5001")

        .. warning::

            - Only supported when Chrome has been started with driverless or the extension at ``selenium_driverless/files/mv3_extension`` has been loaded into the browser.

            - ``Socks5`` doesn't support authentication due to `crbug#1309413 <https://bugs.chromium.org/p/chromium/issues/detail?id=1309413>`__.

        :param username:
        :param password:
        :param host_with_port: in format "example.com:5001"

        """
        # provide auth
        await self._ensure_auth_interception()
        mv3_target = await self.mv3_extension
        arg = {
            "authCredentials": {
                "username": username,
                "password": password
            }
        }
        await mv3_target.execute_script("globalThis.authCreds[arguments[1]] = arguments[0]", arg, host_with_port,
                                        unique_context=False)
        self._auth[host_with_port] = arg

    async def wait_for_cdp(self, event: str, timeout: float or None = None) -> dict:
        """
        wait for an event on the current target
        see :func:`Target.wait_for_cdp <selenium_driverless.types.target.Target.wait_for_cdp>` for reference
        """
        return await self.current_target.wait_for_cdp(event=event, timeout=timeout)

    async def add_cdp_listener(self, event: str, callback: typing.Callable[[dict], any]):
        """
        add a listener for a CDP event on the current target
        see :func:`Target.add_cdp_listener <selenium_driverless.types.target.Target.add_cdp_listener>` for reference
        """
        return await self.current_target.add_cdp_listener(event=event, callback=callback)

    async def remove_cdp_listener(self, event: str, callback: typing.Callable[[dict], any]):
        """
        remove a listener for a CDP event on the current target
        see :func:`Target.remove_cdp_listener <selenium_driverless.types.target.Target.remove_cdp_listener>` for reference
        """
        return await self.current_target.remove_cdp_listener(event=event, callback=callback)

    async def get_cdp_event_iter(self, event: str) -> typing.AsyncIterable[dict]:
        """
        iterate over CDP events on the current target
        see :func:`Target.get_cdp_event_iter <selenium_driverless.types.target.Target.get_cdp_event_iter>` for reference
        """
        return await self.current_target.get_cdp_event_iter(event=event)

    async def execute_cdp_cmd(self, cmd: str, cmd_args: dict or None = None,
                              timeout: float or None = 10) -> dict:
        """Execute Chrome Devtools Protocol command on the current target
        executes it on :class:`Target.execute_cdp_cmd <selenium_driverless.types.base_target.BaseTarget>`
        if ``message:'Not allowed'`` received
        see :func:`Target.execute_cdp_cmd <selenium_driverless.types.target.Target.execute_cdp_cmd>` for reference
        """
        return await self.current_context.execute_cdp_cmd(cmd=cmd, cmd_args=cmd_args, timeout=timeout)

    async def fetch(self, *args, **kwargs) -> dict:
        """
        executes a JS ``fetch`` request within the current target
        see :func:`Target.fetch <selenium_driverless.types.target.Target.fetch>` for reference
        """
        return await self.current_target.fetch(*args, **kwargs)

    async def xhr(self, *args, **kwargs) -> dict:
        """
        executes a JS ``XMLHttpRequest`` request within the current target
        see :func:`Target.fetch <selenium_driverless.types.target.Target.fetch>` for reference
        """
        return await self.current_target.xhr(*args, **kwargs)

    # noinspection PyTypeChecker
    async def get_sinks(self) -> list:
        """
        :Returns: A list of sinks available for Cast.
        """
        return await self.current_target.get_sinks()

    async def get_issue_message(self):
        """
        :Returns: An error message when there is any issue in a Cast session.
        """
        return await self.current_target.get_issue_message()

    async def set_sink_to_use(self, sink_name: str) -> dict:
        """Sets a specific sink, using its name, as a Cast session receiver
        target.

        :param sink_name: Name of the sink to use as the target.
        """
        return await self.current_target.set_sink_to_use(sink_name=sink_name)

    async def start_desktop_mirroring(self, sink_name: str) -> dict:
        """Starts a desktop mirroring session on a specific receiver target.

        :param sink_name: Name of the sink to use as the target.
        """
        return await self.current_target.start_desktop_mirroring(sink_name=sink_name)

    async def start_tab_mirroring(self, sink_name: str) -> dict:
        """Starts a tab mirroring session on a specific receiver target.

        :param sink_name: Name of the sink to use as the target.
        """
        return await self.current_target.start_tab_mirroring(sink_name=sink_name)

    async def stop_casting(self, sink_name: str) -> dict:
        """Stops the existing Cast session on a specific receiver target.

        :Args:
         - sink_name: Name of the sink to stop the Cast session.
        """
        return await self.current_target.stop_casting(sink_name=sink_name)