# 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. import logging import socket import string import os import certifi import urllib3 import platform from base64 import b64encode from urllib import parse from selenium import __version__ from .command import Command from .errorhandler import ErrorCode from . import utils LOGGER = logging.getLogger(__name__) class RemoteConnection(object): """A connection with the Remote WebDriver server. Communicates with the server using the WebDriver wire protocol: https://github.com/SeleniumHQ/selenium/wiki/JsonWireProtocol""" browser_name = None _timeout = socket._GLOBAL_DEFAULT_TIMEOUT _ca_certs = certifi.where() @classmethod def get_timeout(cls): """ :Returns: Timeout value in seconds for all http requests made to the Remote Connection """ return None if cls._timeout == socket._GLOBAL_DEFAULT_TIMEOUT else cls._timeout @classmethod def set_timeout(cls, timeout): """ Override the default timeout :Args: - timeout - timeout value for http requests in seconds """ cls._timeout = timeout @classmethod def reset_timeout(cls): """ Reset the http request timeout to socket._GLOBAL_DEFAULT_TIMEOUT """ cls._timeout = socket._GLOBAL_DEFAULT_TIMEOUT @classmethod def get_certificate_bundle_path(cls): """ :Returns: Paths of the .pem encoded certificate to verify connection to command executor """ return cls._ca_certs @classmethod def set_certificate_bundle_path(cls, path): """ Set the path to the certificate bundle to verify connection to command executor. Can also be set to None to disable certificate validation. :Args: - path - path of a .pem encoded certificate chain. """ cls._ca_certs = path @classmethod def get_remote_connection_headers(cls, parsed_url, keep_alive=False): """ Get headers for remote request. :Args: - parsed_url - The parsed url - keep_alive (Boolean) - Is this a keep-alive connection (default: False) """ system = platform.system().lower() if system == "darwin": system = "mac" headers = { 'Accept': 'application/json', 'Content-Type': 'application/json;charset=UTF-8', 'User-Agent': 'selenium/{} (python {})'.format(__version__, system) } if parsed_url.username: base64string = b64encode('{0.username}:{0.password}'.format(parsed_url).encode()) headers.update({ 'Authorization': 'Basic {}'.format(base64string.decode()) }) if keep_alive: headers.update({ 'Connection': 'keep-alive' }) return headers def _get_proxy_url(self): if self._url.startswith('https://'): return os.environ.get('https_proxy', os.environ.get('HTTPS_PROXY')) elif self._url.startswith('http://'): return os.environ.get('http_proxy', os.environ.get('HTTP_PROXY')) def _get_connection_manager(self): pool_manager_init_args = { 'timeout': self._timeout } if self._ca_certs: pool_manager_init_args['cert_reqs'] = 'CERT_REQUIRED' pool_manager_init_args['ca_certs'] = self._ca_certs return urllib3.PoolManager(**pool_manager_init_args) if not self._proxy_url else \ urllib3.ProxyManager(self._proxy_url, **pool_manager_init_args) def __init__(self, remote_server_addr, keep_alive=False, resolve_ip=None, ignore_proxy=False): if resolve_ip: import warnings warnings.warn( "'resolve_ip' option removed; ip addresses are now always resolved by urllib3.", DeprecationWarning) self.keep_alive = keep_alive self._url = remote_server_addr # Env var NO_PROXY will override this part of the code _no_proxy = os.environ.get('no_proxy', os.environ.get('NO_PROXY')) if _no_proxy: for npu in _no_proxy.split(','): npu = npu.strip() if npu == "*": ignore_proxy = True break n_url = parse.urlparse(npu) remote_add = parse.urlparse(self._url) if n_url.netloc: if remote_add.netloc == n_url.netloc: ignore_proxy = True break else: if n_url.path in remote_add.netloc: ignore_proxy = True break self._proxy_url = self._get_proxy_url() if not ignore_proxy else None if keep_alive: self._conn = self._get_connection_manager() self._commands = { Command.STATUS: ('GET', '/status'), Command.NEW_SESSION: ('POST', '/session'), Command.GET_ALL_SESSIONS: ('GET', '/sessions'), Command.QUIT: ('DELETE', '/session/$sessionId'), Command.GET_CURRENT_WINDOW_HANDLE: ('GET', '/session/$sessionId/window_handle'), Command.W3C_GET_CURRENT_WINDOW_HANDLE: ('GET', '/session/$sessionId/window'), Command.GET_WINDOW_HANDLES: ('GET', '/session/$sessionId/window_handles'), Command.W3C_GET_WINDOW_HANDLES: ('GET', '/session/$sessionId/window/handles'), Command.GET: ('POST', '/session/$sessionId/url'), Command.GO_FORWARD: ('POST', '/session/$sessionId/forward'), Command.GO_BACK: ('POST', '/session/$sessionId/back'), Command.REFRESH: ('POST', '/session/$sessionId/refresh'), Command.EXECUTE_SCRIPT: ('POST', '/session/$sessionId/execute'), Command.W3C_EXECUTE_SCRIPT: ('POST', '/session/$sessionId/execute/sync'), Command.W3C_EXECUTE_SCRIPT_ASYNC: ('POST', '/session/$sessionId/execute/async'), Command.GET_CURRENT_URL: ('GET', '/session/$sessionId/url'), Command.GET_TITLE: ('GET', '/session/$sessionId/title'), Command.GET_PAGE_SOURCE: ('GET', '/session/$sessionId/source'), Command.SCREENSHOT: ('GET', '/session/$sessionId/screenshot'), Command.ELEMENT_SCREENSHOT: ('GET', '/session/$sessionId/element/$id/screenshot'), Command.FIND_ELEMENT: ('POST', '/session/$sessionId/element'), Command.FIND_ELEMENTS: ('POST', '/session/$sessionId/elements'), Command.W3C_GET_ACTIVE_ELEMENT: ('GET', '/session/$sessionId/element/active'), Command.GET_ACTIVE_ELEMENT: ('POST', '/session/$sessionId/element/active'), Command.FIND_CHILD_ELEMENT: ('POST', '/session/$sessionId/element/$id/element'), Command.FIND_CHILD_ELEMENTS: ('POST', '/session/$sessionId/element/$id/elements'), Command.CLICK_ELEMENT: ('POST', '/session/$sessionId/element/$id/click'), Command.CLEAR_ELEMENT: ('POST', '/session/$sessionId/element/$id/clear'), Command.SUBMIT_ELEMENT: ('POST', '/session/$sessionId/element/$id/submit'), Command.GET_ELEMENT_TEXT: ('GET', '/session/$sessionId/element/$id/text'), Command.SEND_KEYS_TO_ELEMENT: ('POST', '/session/$sessionId/element/$id/value'), Command.SEND_KEYS_TO_ACTIVE_ELEMENT: ('POST', '/session/$sessionId/keys'), Command.UPLOAD_FILE: ('POST', "/session/$sessionId/se/file"), Command.GET_ELEMENT_VALUE: ('GET', '/session/$sessionId/element/$id/value'), Command.GET_ELEMENT_TAG_NAME: ('GET', '/session/$sessionId/element/$id/name'), Command.IS_ELEMENT_SELECTED: ('GET', '/session/$sessionId/element/$id/selected'), Command.SET_ELEMENT_SELECTED: ('POST', '/session/$sessionId/element/$id/selected'), Command.IS_ELEMENT_ENABLED: ('GET', '/session/$sessionId/element/$id/enabled'), Command.IS_ELEMENT_DISPLAYED: ('GET', '/session/$sessionId/element/$id/displayed'), Command.GET_ELEMENT_LOCATION: ('GET', '/session/$sessionId/element/$id/location'), Command.GET_ELEMENT_LOCATION_ONCE_SCROLLED_INTO_VIEW: ('GET', '/session/$sessionId/element/$id/location_in_view'), Command.GET_ELEMENT_SIZE: ('GET', '/session/$sessionId/element/$id/size'), Command.GET_ELEMENT_RECT: ('GET', '/session/$sessionId/element/$id/rect'), Command.GET_ELEMENT_ATTRIBUTE: ('GET', '/session/$sessionId/element/$id/attribute/$name'), Command.GET_ELEMENT_PROPERTY: ('GET', '/session/$sessionId/element/$id/property/$name'), Command.GET_ELEMENT_ARIA_ROLE: ('GET', '/session/$sessionId/element/$id/computedrole'), Command.GET_ELEMENT_ARIA_LABEL: ('GET', '/session/$sessionId/element/$id/computedlabel'), Command.GET_SHADOW_ROOT: ('GET', '/session/$sessionId/element/$id/shadow'), Command.FIND_ELEMENT_FROM_SHADOW_ROOT: ('POST', '/session/$sessionId/shadow/$shadowId/element'), Command.FIND_ELEMENTS_FROM_SHADOW_ROOT: ('POST', '/session/$sessionId/shadow/$shadowId/elements'), Command.GET_ALL_COOKIES: ('GET', '/session/$sessionId/cookie'), Command.ADD_COOKIE: ('POST', '/session/$sessionId/cookie'), Command.GET_COOKIE: ('GET', '/session/$sessionId/cookie/$name'), Command.DELETE_ALL_COOKIES: ('DELETE', '/session/$sessionId/cookie'), Command.DELETE_COOKIE: ('DELETE', '/session/$sessionId/cookie/$name'), Command.SWITCH_TO_FRAME: ('POST', '/session/$sessionId/frame'), Command.SWITCH_TO_PARENT_FRAME: ('POST', '/session/$sessionId/frame/parent'), Command.SWITCH_TO_WINDOW: ('POST', '/session/$sessionId/window'), Command.NEW_WINDOW: ('POST', '/session/$sessionId/window/new'), Command.CLOSE: ('DELETE', '/session/$sessionId/window'), Command.GET_ELEMENT_VALUE_OF_CSS_PROPERTY: ('GET', '/session/$sessionId/element/$id/css/$propertyName'), Command.IMPLICIT_WAIT: ('POST', '/session/$sessionId/timeouts/implicit_wait'), Command.EXECUTE_ASYNC_SCRIPT: ('POST', '/session/$sessionId/execute_async'), Command.SET_SCRIPT_TIMEOUT: ('POST', '/session/$sessionId/timeouts/async_script'), Command.SET_TIMEOUTS: ('POST', '/session/$sessionId/timeouts'), Command.GET_TIMEOUTS: ('GET', '/session/$sessionId/timeouts'), Command.DISMISS_ALERT: ('POST', '/session/$sessionId/dismiss_alert'), Command.W3C_DISMISS_ALERT: ('POST', '/session/$sessionId/alert/dismiss'), Command.ACCEPT_ALERT: ('POST', '/session/$sessionId/accept_alert'), Command.W3C_ACCEPT_ALERT: ('POST', '/session/$sessionId/alert/accept'), Command.SET_ALERT_VALUE: ('POST', '/session/$sessionId/alert_text'), Command.W3C_SET_ALERT_VALUE: ('POST', '/session/$sessionId/alert/text'), Command.GET_ALERT_TEXT: ('GET', '/session/$sessionId/alert_text'), Command.W3C_GET_ALERT_TEXT: ('GET', '/session/$sessionId/alert/text'), Command.SET_ALERT_CREDENTIALS: ('POST', '/session/$sessionId/alert/credentials'), Command.CLICK: ('POST', '/session/$sessionId/click'), Command.W3C_ACTIONS: ('POST', '/session/$sessionId/actions'), Command.W3C_CLEAR_ACTIONS: ('DELETE', '/session/$sessionId/actions'), Command.DOUBLE_CLICK: ('POST', '/session/$sessionId/doubleclick'), Command.MOUSE_DOWN: ('POST', '/session/$sessionId/buttondown'), Command.MOUSE_UP: ('POST', '/session/$sessionId/buttonup'), Command.MOVE_TO: ('POST', '/session/$sessionId/moveto'), Command.GET_WINDOW_SIZE: ('GET', '/session/$sessionId/window/$windowHandle/size'), Command.SET_WINDOW_SIZE: ('POST', '/session/$sessionId/window/$windowHandle/size'), Command.GET_WINDOW_POSITION: ('GET', '/session/$sessionId/window/$windowHandle/position'), Command.SET_WINDOW_POSITION: ('POST', '/session/$sessionId/window/$windowHandle/position'), Command.SET_WINDOW_RECT: ('POST', '/session/$sessionId/window/rect'), Command.GET_WINDOW_RECT: ('GET', '/session/$sessionId/window/rect'), Command.MAXIMIZE_WINDOW: ('POST', '/session/$sessionId/window/$windowHandle/maximize'), Command.W3C_MAXIMIZE_WINDOW: ('POST', '/session/$sessionId/window/maximize'), Command.SET_SCREEN_ORIENTATION: ('POST', '/session/$sessionId/orientation'), Command.GET_SCREEN_ORIENTATION: ('GET', '/session/$sessionId/orientation'), Command.SINGLE_TAP: ('POST', '/session/$sessionId/touch/click'), Command.TOUCH_DOWN: ('POST', '/session/$sessionId/touch/down'), Command.TOUCH_UP: ('POST', '/session/$sessionId/touch/up'), Command.TOUCH_MOVE: ('POST', '/session/$sessionId/touch/move'), Command.TOUCH_SCROLL: ('POST', '/session/$sessionId/touch/scroll'), Command.DOUBLE_TAP: ('POST', '/session/$sessionId/touch/doubleclick'), Command.LONG_PRESS: ('POST', '/session/$sessionId/touch/longclick'), Command.FLICK: ('POST', '/session/$sessionId/touch/flick'), Command.EXECUTE_SQL: ('POST', '/session/$sessionId/execute_sql'), Command.GET_LOCATION: ('GET', '/session/$sessionId/location'), Command.SET_LOCATION: ('POST', '/session/$sessionId/location'), Command.GET_APP_CACHE: ('GET', '/session/$sessionId/application_cache'), Command.GET_APP_CACHE_STATUS: ('GET', '/session/$sessionId/application_cache/status'), Command.CLEAR_APP_CACHE: ('DELETE', '/session/$sessionId/application_cache/clear'), Command.GET_NETWORK_CONNECTION: ('GET', '/session/$sessionId/network_connection'), Command.SET_NETWORK_CONNECTION: ('POST', '/session/$sessionId/network_connection'), Command.GET_LOCAL_STORAGE_ITEM: ('GET', '/session/$sessionId/local_storage/key/$key'), Command.REMOVE_LOCAL_STORAGE_ITEM: ('DELETE', '/session/$sessionId/local_storage/key/$key'), Command.GET_LOCAL_STORAGE_KEYS: ('GET', '/session/$sessionId/local_storage'), Command.SET_LOCAL_STORAGE_ITEM: ('POST', '/session/$sessionId/local_storage'), Command.CLEAR_LOCAL_STORAGE: ('DELETE', '/session/$sessionId/local_storage'), Command.GET_LOCAL_STORAGE_SIZE: ('GET', '/session/$sessionId/local_storage/size'), Command.GET_SESSION_STORAGE_ITEM: ('GET', '/session/$sessionId/session_storage/key/$key'), Command.REMOVE_SESSION_STORAGE_ITEM: ('DELETE', '/session/$sessionId/session_storage/key/$key'), Command.GET_SESSION_STORAGE_KEYS: ('GET', '/session/$sessionId/session_storage'), Command.SET_SESSION_STORAGE_ITEM: ('POST', '/session/$sessionId/session_storage'), Command.CLEAR_SESSION_STORAGE: ('DELETE', '/session/$sessionId/session_storage'), Command.GET_SESSION_STORAGE_SIZE: ('GET', '/session/$sessionId/session_storage/size'), Command.GET_LOG: ('POST', '/session/$sessionId/se/log'), Command.GET_AVAILABLE_LOG_TYPES: ('GET', '/session/$sessionId/se/log/types'), Command.CURRENT_CONTEXT_HANDLE: ('GET', '/session/$sessionId/context'), Command.CONTEXT_HANDLES: ('GET', '/session/$sessionId/contexts'), Command.SWITCH_TO_CONTEXT: ('POST', '/session/$sessionId/context'), Command.FULLSCREEN_WINDOW: ('POST', '/session/$sessionId/window/fullscreen'), Command.MINIMIZE_WINDOW: ('POST', '/session/$sessionId/window/minimize'), Command.PRINT_PAGE: ('POST', '/session/$sessionId/print') } def execute(self, command, params): """ Send a command to the remote server. Any path substitutions required for the URL mapped to the command should be included in the command parameters. :Args: - command - A string specifying the command to execute. - params - A dictionary of named parameters to send with the command as its JSON payload. """ command_info = self._commands[command] assert command_info is not None, 'Unrecognised command %s' % command path = string.Template(command_info[1]).substitute(params) if isinstance(params, dict) and 'sessionId' in params: del params['sessionId'] data = utils.dump_json(params) url = f"{self._url}{path}" return self._request(command_info[0], url, body=data) def _request(self, method, url, body=None): """ Send an HTTP request to the remote server. :Args: - method - A string for the HTTP method to send the request with. - url - A string for the URL to send the request to. - body - A string for request body. Ignored unless method is POST or PUT. :Returns: A dictionary with the server's parsed JSON response. """ LOGGER.debug(f"{method} {url} {body}") parsed_url = parse.urlparse(url) headers = self.get_remote_connection_headers(parsed_url, self.keep_alive) resp = None if body and method != 'POST' and method != 'PUT': body = None if self.keep_alive: resp = self._conn.request(method, url, body=body, headers=headers) statuscode = resp.status else: conn = self._get_connection_manager() with conn as http: resp = http.request(method, url, body=body, headers=headers) statuscode = resp.status if not hasattr(resp, 'getheader'): if hasattr(resp.headers, 'getheader'): resp.getheader = lambda x: resp.headers.getheader(x) elif hasattr(resp.headers, 'get'): resp.getheader = lambda x: resp.headers.get(x) data = resp.data.decode('UTF-8') try: if 300 <= statuscode < 304: return self._request('GET', resp.getheader('location')) if 399 < statuscode <= 500: return {'status': statuscode, 'value': data} content_type = [] if resp.getheader('Content-Type'): content_type = resp.getheader('Content-Type').split(';') if not any([x.startswith('image/png') for x in content_type]): try: data = utils.load_json(data.strip()) except ValueError: if 199 < statuscode < 300: status = ErrorCode.SUCCESS else: status = ErrorCode.UNKNOWN_ERROR return {'status': status, 'value': data.strip()} # Some of the drivers incorrectly return a response # with no 'value' field when they should return null. if 'value' not in data: data['value'] = None return data else: data = {'status': 0, 'value': data} return data finally: LOGGER.debug("Finished Request") resp.close() def close(self): """ Clean up resources when finished with the remote_connection """ if hasattr(self, '_conn'): self._conn.clear()