File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/http.tar
Back
__pycache__/__init__.cpython-310.pyc 0000644 00000022363 15030123507 0013216 0 ustar 00 o �ϴfm- � @ s d dl Z d dlZd dlZd dlZd dlZd dlmZmZmZm Z m Z d dl m Z m Z d dlmZmZ d dlmZmZmZ dZdZdZd Zd Ze�e�e ��Z e d d e fd ee e f fde fg�Z e dd e fd ee e f fde fdee ef fdee fg�Z!de de"fdd�Z#de de e de de e fdd�Z$i a%de e d e e ddfd d �Z&dee e f fd d �Z' d3d!e j(d"e e de fd#d$�Z)d%d&� Z*d'd(� Z+ d4d!e j(d"e e d e e de fd)d*�Z,de e fd+d,�Z-di ddd-fde d.e e. d ee e f d/e e d"e e d0e"de!fd1d2�Z/dS )5� N)�Any�Dict�List� NamedTuple�Optional)�error�request)� ParseResult�urlparse)�defaults� exceptions�util)z169.254.169.254�metadataz[fd00:ec2::254]zhttp://archive.ubuntu.comzhttps://esm.ubuntu.comzhttp://api.snapcraft.iozhttps://api.snapcraft.io�UnparsedHTTPResponse�code�headers�body� HTTPResponse� json_dict� json_list�url�returnc C sR zt | �}W n ty Y dS w |jdvrdS z|j W dS ty( Y dS w )NF)�https�httpT)r � ValueError�scheme�port)r � parsed_url� r �8/usr/lib/python3/dist-packages/uaclient/http/__init__.py�is_service_url( s � ��r �protocol�proxy�test_urlc C sF |sd S t |�stj|d��tj|dd�}| dkrjt|�jdkrjzt||d�}W n4 tjy1 � tj y9 � tj yA � t y\ } zt � d||t|�� tj|d��d }~ww |jdkrd|S tj|d��t�| |i�}t�|�}z|�|� |W S tjt jfy� } zt � d||t|d t|��� tj|d��d }~ww ) N)r" �HEAD)�methodr �� https_proxyz:Error trying to use "%s" as pycurl proxy to reach "%s": %s�� z:Error trying to use "%s" as urllib proxy to reach "%s": %s�reason)r r �ProxyInvalidUrlr �Requestr r � _readurl_pycurl_https_in_https�PycurlRequiredError�ProxyAuthenticationFailed�PycurlCACertificatesError� Exception�LOGr �str�ProxyNotWorkingErrorr � ProxyHandler� build_opener�open�socket�timeout�URLError�getattr)r! r" r# �req�response�e� proxy_handler�openerr r r �validate_proxy9 sT � �� � ��r@ � http_proxyr' c C s� i }| r| |d< |r||d< d� tt��}dD ]}tj�|�}|r2d� tt|�d���tt����}qt � d|� |tjd< |tjd< |rTt � |�}t � |�}t �|� t j dd |id � |ad S ) aW Globally configure pro-client to use http and https proxies. - sets global proxy configuration for urllib - sets the no_proxy environment variable for the current process which gets inherited for all subprocesses - sets module variable for use in https-in-https pycurl requests this is retrieved later using get_configured_web_proxy :param http_proxy: http proxy to be used by urllib. If None, it will not be configured :param https_proxy: https proxy to be used by urllib. If None, it will not be configured r r �,)�no_proxy�NO_PROXYzSetting no_proxy: %srC rD zSetting global proxy dict�extra)rE N)�join�sorted�UA_NO_PROXY_URLS�os�environ�get�set�split�unionr1 �debugr r4 r5 �install_opener�_global_proxy_dict)rA r' � proxy_dictrC �env_var� proxy_valuer> r? r r r �configure_web_proxyn s0 ��� rU c C s t S �N)rQ r r r r �get_configured_web_proxy� s rW r; r8 c C s� z t j| |d�}W n1 tjy } z|}W Y d }~n!d }~w tjy: } zt�t|j�� t j || j d��d }~ww |� � � d�}dd� |j�� D �}t|j||d�S )N�r8 )�causer �utf-8c S s i | ] \}}|� � |�qS r )�lower)�.0�k�vr r r � <dictcomp>� s z#_readurl_urllib.<locals>.<dictcomp>�r r r )r �urlopenr � HTTPErrorr9 r1 � exceptionr2 r) r �ConnectivityError�full_url�read�decoder �itemsr r )r; r8 �respr= r r r r r �_readurl_urllib� s" ����rj c C sJ t |�}t| �}|jdko t�|j� o |duo |jdk}t�d|� |S )a� We only want to use pycurl if all of the following are true - The target url scheme is https - The target host is not in no_proxy - An https_proxy is configured either via pro's config or via environment - The https_proxy url scheme is https urllib.request provides some helpful functions that we re-use here. This function also returns the https_proxy to use, since it is calculated here anyway. r NzShould use pycurl: %r)r �_parse_https_proxyr r � proxy_bypass�hostnamer1 rO )r' � target_url�parsed_target_url�parsed_https_proxy�retr r r �should_use_pycurl� s ��� rr c C st d }d }t | j�dkr| jd }t | j�dkr | jd }||kr*|r*d|v r*t�� �||kr4tj|d��tj| d��)Nr � �407�r )r= )�len�argsr r. r/ � PycurlError)r r �authentication_error_code�ca_certificates_error_coder �msgr r r �_handle_pycurl_error� s r| c s zdd l }W n ty t�� �w |�� }| �� �� }|dkr(|�|jd� n*|dkr4|�|j d� n |dkrK|�|j d� | j rJ|�|j | j � nt d�|���|�|j| �� � dd� | �� D �}t|�dkrq|�|j|� |�|jd� |�|jtj� |r�|�|j|� |r�t|�}|r�|�� nd }|�|j|� |�|j d � nt � d � t � � }|�|j!|� i � � fd d �} |�|j"| � z|�#� W n |j$y� } zt%| | �� |j&|j'd � W Y d } ~ nd } ~ ww t(|�)|j*��} |�+� �,d�} |�-� t.| � | d�S )Nr �GETTr$ �POSTz5HTTP method "{}" not supported in HTTPS-in-HTTPS modec S s g | ] \}}d � ||��qS )z{}: {}��format)r\ �name�valr r r � <listcomp> s �z2_readurl_pycurl_https_in_https.<locals>.<listcomp>� z1in pycurl request function without an https proxyc sF | � d�} d| vr d S | �dd�\}}|�� �� }|�� }|� |< d S )Nz iso-8859-1�:rs )rg rM �stripr[ )� header_line�name_raw� value_rawr� �value�r r r � save_header" s z3_readurl_pycurl_https_in_https.<locals>.save_header)r ry rz rZ r` )/�pycurl� ImportErrorr r- �Curl� get_method�upper�setopt�HTTPGET�NOBODYr~ �data�COPYPOSTFIELDSr r� �URL� get_full_url� header_itemsrv � HTTPHEADER�FOLLOWLOCATION�CAINFOr �SSL_CERTS_PATH�TIMEOUTrk �geturl�PROXY� PROXYTYPEr1 �warning�io�BytesIO� WRITEDATA�HEADERFUNCTION�performr r| � E_RECV_ERROR�E_SSL_CACERT_BADFILE�int�getinfo� RESPONSE_CODE�getvaluerg �closer ) r; r8 r' r� �cr% �header_str_listrp � body_outputr� r= r r r r� r r, � s| � ���� � ����r, c C s" | s t �� �d�} | rt| �S d S )Nr )r � getproxiesrK r r& r r r rk D s rk Tr� r% �log_response_bodyc sn t | �s tj| d��|r|sd}tj| |� |d�}d�� fdd�t� �D ��}t�d� |p.d| ||r7|� d �nd �� t � � d �}t || �rNt|||d ��nt||d ��i } g } d �j� dd�v rytj�jtjd�} t| t�rr| } nt| t�ry| } d��fdd�t�j�D ��}d� |p�d| |�} |r��j} | r�| } n| r�| } | d� | �7 } t�| � t�j�j�j| | d�S )Nru r~ )r� r r% z, c s g | ] }d � |� | ��qS �z '{}': '{}'r �r\ r] r� r r r� Z s zreadurl.<locals>.<listcomp>z'URL [{}]: {}, headers: {{{}}}, data: {}r} rZ r )r8 r' rX zapplication/jsonz content-type� )�clsc s g | ] }d � |� j| ��qS r� )r� r r� )ri r r r� w s z&URL [{}] response: {}, headers: {{{}}}z , data: {})r r r r r )r r � InvalidUrlr r+ rF rG r1 rO r� rg rW rK rr r, rj r �json�loadsr r �DatetimeAwareJSONDecoder� isinstance�dict�listr r )r r� r r% r8 r� r; �sorted_header_strr' r r � json_body� debug_msg� body_to_logr )r ri r �readurlJ sf ��� � � � �r� rV )NN)0r� r� �loggingrI r7 �typingr r r r r �urllibr r � urllib.parser r �uaclientr r r rH � PROXY_VALIDATION_APT_HTTP_URL� PROXY_VALIDATION_APT_HTTPS_URL� PROXY_VALIDATION_SNAP_HTTP_URL� PROXY_VALIDATION_SNAP_HTTPS_URL� getLogger� replace_top_level_logger_name�__name__r1 r� r2 r r �boolr r@ rQ rU rW r+ rj rr r| r, rk �bytesr� r r r r �<module> s� �� �� ��� �2�� �1��� ����� �_��� ����� __pycache__/serviceclient.cpython-310.pyc 0000644 00000010121 15030123507 0014303 0 ustar 00 o �ϴf� � @ st d dl Z d dlZd dlZd dlZd dlmZmZmZ d dlm Z d dl m Z m Z m Z mZmZ G dd� de jd�ZdS )� N)�Any�Dict�Optional)� urlencode)�config�http�system�util�versionc @ s� e Zd ZdZdZeejdefdd���Z dde e j ddfdd�Z d d � Z dd ed e e dejfdd�Zdefdd�Zdede ej fdd�ZdS )�UAServiceClient� N�returnc C s dS )z@String in subclasses, the UAConfig attribute containing base urlN� ��selfr r �=/usr/lib/python3/dist-packages/uaclient/http/serviceclient.py�cfg_url_base_attr s z!UAServiceClient.cfg_url_base_attr�cfgc C s |s t �� | _d S || _d S �N)r �UAConfigr )r r r r r �__init__ s zUAServiceClient.__init__c C s d� t�� �ddd�S )Nz UA-Client/{}�application/json)z user-agent�accept� content-type)�formatr � get_versionr r r r �headers s �zUAServiceClient.headersT�log_response_body�timeoutc C s� |� d�}|s | �� }|�d�dkr |r tj|tjd��d�}t� t | j | j �|�}| � |�} | r3| S |rHdd� t|�� �D �} |dt| � 7 }|d urN|n| j} tj||||| |d �S ) N�/r r )�clszutf-8c S s i | ] \}}|d ur||�qS r r )�.0�k�vr r r � <dictcomp>> s z/UAServiceClient.request_url.<locals>.<dictcomp>�?)�url�datar �methodr r )�lstripr �get�json�dumpsr �DatetimeAwareJSONEncoder�encode� posixpath�join�getattrr r �_get_fake_responses�sorted�itemsr � url_timeoutr �readurl) r �pathr' r r( � query_paramsr r r& � fake_response�filtered_params�timeout_to_user r r � request_url% s2 � ��zUAServiceClient.request_urlr& c C sf | j dur | j �|g �S | jj�d�}|si | _ ntj�|�s#i | _ n t�t � |��| _ | j �|g �S )a. Return a list of fake response dicts for a given URL. serviceclient_url_responses in uaclient.conf should be a path to a json file which contains a dictionary keyed by full URL path. Each value will be a list of dicts representing each faked response for the given URL. The response dict item will have a code: <HTTP_STATUS_CODE> and response: "some string of content". The JSON string below fakes the available_resources URL on the contract server: '{"https://contracts.canonical.com/v1/resources": [{"code": 200, "response": {"key": "val1", "key2": "val2"}}]}' :return: List of dicts for each faked response matching the url, or and empty list when no matching url found. N�serviceclient_url_responses) �_response_overlayr* r �features�osr7 �existsr+ �loadsr � load_file)r r& �response_overlay_pathr r r �_get_response_overlayM s � �z%UAServiceClient._get_response_overlayc C s� | � |�}|s dS t|�dkr|d }n|�d�}i }g }|d }t|t�r)|}nt|t�r0|}tj|d |�di �t j |d dd�||d �S ) z7Return response if faked for this URL in uaclient.conf.N� r �response�coder T)� sort_keys)rH r �body� json_dict� json_list) rE �len�pop� isinstance�dict�listr � HTTPResponser* r+ r, )r r& � responsesrG rK rL �respr r r r2 o s( �z#UAServiceClient._get_fake_responsesr )NNNNTN)�__name__� __module__� __qualname__r5 r> �property�abc�abstractmethod�strr r r r r r �bool�intr rR r< rE r2 r r r r r s. ��� �( "r )� metaclass)rY r+ r@ r/ �typingr r r � urllib.parser �uaclientr r r r r �ABCMetar r r r r �<module> s serviceclient.py 0000644 00000011352 15030123507 0007753 0 ustar 00 import abc import json import os import posixpath from typing import Any, Dict, Optional # noqa: F401 from urllib.parse import urlencode from uaclient import config, http, system, util, version class UAServiceClient(metaclass=abc.ABCMeta): url_timeout = 30 # type: Optional[int] # Cached serviceclient_url_responses if provided in uaclient.conf # via features: {serviceclient_url_responses: /some/file.json} _response_overlay = None # type: Dict[str, Any] @property @abc.abstractmethod def cfg_url_base_attr(self) -> str: """String in subclasses, the UAConfig attribute containing base url""" pass def __init__(self, cfg: Optional[config.UAConfig] = None) -> None: if not cfg: self.cfg = config.UAConfig() else: self.cfg = cfg def headers(self): return { "user-agent": "UA-Client/{}".format(version.get_version()), "accept": "application/json", "content-type": "application/json", } def request_url( self, path, data=None, headers=None, method=None, query_params=None, log_response_body: bool = True, timeout: Optional[int] = None, ) -> http.HTTPResponse: path = path.lstrip("/") if not headers: headers = self.headers() if headers.get("content-type") == "application/json" and data: data = json.dumps(data, cls=util.DatetimeAwareJSONEncoder).encode( "utf-8" ) url = posixpath.join(getattr(self.cfg, self.cfg_url_base_attr), path) fake_response = self._get_fake_responses(url) if fake_response: return fake_response # URL faked by uaclient.conf if query_params: # filter out None values filtered_params = { k: v for k, v in sorted(query_params.items()) if v is not None } url += "?" + urlencode(filtered_params) timeout_to_use = timeout if timeout is not None else self.url_timeout return http.readurl( url=url, data=data, headers=headers, method=method, timeout=timeout_to_use, log_response_body=log_response_body, ) def _get_response_overlay(self, url: str): """Return a list of fake response dicts for a given URL. serviceclient_url_responses in uaclient.conf should be a path to a json file which contains a dictionary keyed by full URL path. Each value will be a list of dicts representing each faked response for the given URL. The response dict item will have a code: <HTTP_STATUS_CODE> and response: "some string of content". The JSON string below fakes the available_resources URL on the contract server: '{"https://contracts.canonical.com/v1/resources": \ [{"code": 200, "response": {"key": "val1", "key2": "val2"}}]}' :return: List of dicts for each faked response matching the url, or and empty list when no matching url found. """ if self._response_overlay is not None: # Cache it so we don't re-read config every readurl call return self._response_overlay.get(url, []) response_overlay_path = self.cfg.features.get( "serviceclient_url_responses" ) if not response_overlay_path: self._response_overlay = {} elif not os.path.exists(response_overlay_path): self._response_overlay = {} else: self._response_overlay = json.loads( system.load_file(response_overlay_path) ) return self._response_overlay.get(url, []) def _get_fake_responses(self, url: str) -> Optional[http.HTTPResponse]: """Return response if faked for this URL in uaclient.conf.""" responses = self._get_response_overlay(url) if not responses: return None if len(responses) == 1: # When only one response is defined, repeat it for all calls response = responses[0] else: # When multiple responses defined pop the first one off the list. response = responses.pop(0) json_dict = {} json_list = [] resp = response["response"] if isinstance(resp, dict): json_dict = resp elif isinstance(resp, list): json_list = resp return http.HTTPResponse( code=response["code"], headers=response.get("headers", {}), body=json.dumps(response["response"], sort_keys=True), json_dict=json_dict, json_list=json_list, ) __init__.py 0000644 00000026555 15030123507 0006666 0 ustar 00 import io import json import logging import os import socket from typing import Any, Dict, List, NamedTuple, Optional from urllib import error, request from urllib.parse import ParseResult, urlparse from uaclient import defaults, exceptions, util UA_NO_PROXY_URLS = ("169.254.169.254", "metadata", "[fd00:ec2::254]") PROXY_VALIDATION_APT_HTTP_URL = "http://archive.ubuntu.com" PROXY_VALIDATION_APT_HTTPS_URL = "https://esm.ubuntu.com" PROXY_VALIDATION_SNAP_HTTP_URL = "http://api.snapcraft.io" PROXY_VALIDATION_SNAP_HTTPS_URL = "https://api.snapcraft.io" LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) UnparsedHTTPResponse = NamedTuple( "UnparsedHTTPResponse", [ ("code", int), ("headers", Dict[str, str]), ("body", str), ], ) HTTPResponse = NamedTuple( "HTTPResponse", [ ("code", int), ("headers", Dict[str, str]), ("body", str), ("json_dict", Dict[str, Any]), ("json_list", List[Any]), ], ) def is_service_url(url: str) -> bool: try: parsed_url = urlparse(url) except ValueError: return False if parsed_url.scheme not in ("https", "http"): return False try: parsed_url.port except ValueError: return False return True def validate_proxy( protocol: str, proxy: Optional[str], test_url: str ) -> Optional[str]: if not proxy: return None if not is_service_url(proxy): raise exceptions.ProxyInvalidUrl(proxy=proxy) req = request.Request(test_url, method="HEAD") if protocol == "https" and urlparse(proxy).scheme == "https": try: response = _readurl_pycurl_https_in_https(req, https_proxy=proxy) except exceptions.PycurlRequiredError: raise except exceptions.ProxyAuthenticationFailed: raise except exceptions.PycurlCACertificatesError: raise except Exception as e: LOG.error( 'Error trying to use "%s" as pycurl proxy to reach "%s": %s', proxy, test_url, str(e), ) raise exceptions.ProxyNotWorkingError(proxy=proxy) if response.code == 200: return proxy else: raise exceptions.ProxyNotWorkingError(proxy=proxy) proxy_handler = request.ProxyHandler({protocol: proxy}) opener = request.build_opener(proxy_handler) try: opener.open(req) return proxy except (socket.timeout, error.URLError) as e: LOG.error( 'Error trying to use "%s" as urllib proxy to reach "%s": %s', proxy, test_url, getattr(e, "reason", str(e)), ) raise exceptions.ProxyNotWorkingError(proxy=proxy) _global_proxy_dict = {} def configure_web_proxy( http_proxy: Optional[str], https_proxy: Optional[str] ) -> None: """ Globally configure pro-client to use http and https proxies. - sets global proxy configuration for urllib - sets the no_proxy environment variable for the current process which gets inherited for all subprocesses - sets module variable for use in https-in-https pycurl requests this is retrieved later using get_configured_web_proxy :param http_proxy: http proxy to be used by urllib. If None, it will not be configured :param https_proxy: https proxy to be used by urllib. If None, it will not be configured """ proxy_dict = {} if http_proxy: proxy_dict["http"] = http_proxy if https_proxy: proxy_dict["https"] = https_proxy # Default no_proxy if absense of NO_PROXY, no_proxy environment vars. no_proxy = ",".join(sorted(UA_NO_PROXY_URLS)) for env_var in ("no_proxy", "NO_PROXY"): proxy_value = os.environ.get(env_var) if proxy_value: # Honor no proxy values and extend UA-specific where absent no_proxy = ",".join( sorted( set(proxy_value.split(",")).union(set(UA_NO_PROXY_URLS)) ) ) LOG.debug("Setting no_proxy: %s", no_proxy) os.environ["no_proxy"] = no_proxy os.environ["NO_PROXY"] = no_proxy if proxy_dict: proxy_handler = request.ProxyHandler(proxy_dict) opener = request.build_opener(proxy_handler) request.install_opener(opener) LOG.debug("Setting global proxy dict", extra={"extra": proxy_dict}) global _global_proxy_dict _global_proxy_dict = proxy_dict def get_configured_web_proxy() -> Dict[str, str]: return _global_proxy_dict def _readurl_urllib( req: request.Request, timeout: Optional[int] = None ) -> UnparsedHTTPResponse: try: resp = request.urlopen(req, timeout=timeout) # nosec B310 except error.HTTPError as e: resp = e except error.URLError as e: LOG.exception(str(e.reason)) raise exceptions.ConnectivityError(cause=e, url=req.full_url) body = resp.read().decode("utf-8") # convert EmailMessage header object to dict with lowercase keys headers = {k.lower(): v for k, v, in resp.headers.items()} return UnparsedHTTPResponse( code=resp.code, headers=headers, body=body, ) def should_use_pycurl(https_proxy, target_url): """ We only want to use pycurl if all of the following are true - The target url scheme is https - The target host is not in no_proxy - An https_proxy is configured either via pro's config or via environment - The https_proxy url scheme is https urllib.request provides some helpful functions that we re-use here. This function also returns the https_proxy to use, since it is calculated here anyway. """ parsed_target_url = urlparse(target_url) parsed_https_proxy = _parse_https_proxy(https_proxy) ret = ( parsed_target_url.scheme == "https" and not request.proxy_bypass(parsed_target_url.hostname) and parsed_https_proxy is not None and parsed_https_proxy.scheme == "https" ) LOG.debug("Should use pycurl: %r", ret) return ret def _handle_pycurl_error( error, url, authentication_error_code, ca_certificates_error_code ): code = None msg = None if len(error.args) > 0: code = error.args[0] if len(error.args) > 1: msg = error.args[1] if code == authentication_error_code and msg and "407" in msg: raise exceptions.ProxyAuthenticationFailed() elif code == ca_certificates_error_code: raise exceptions.PycurlCACertificatesError(url=url) else: raise exceptions.PycurlError(e=error) def _readurl_pycurl_https_in_https( req: request.Request, timeout: Optional[int] = None, https_proxy: Optional[str] = None, ) -> UnparsedHTTPResponse: try: import pycurl except ImportError: raise exceptions.PycurlRequiredError() c = pycurl.Curl() # Method method = req.get_method().upper() if method == "GET": c.setopt(pycurl.HTTPGET, True) elif method == "HEAD": c.setopt(pycurl.NOBODY, True) elif method == "POST": c.setopt(pycurl.POST, True) if req.data: c.setopt(pycurl.COPYPOSTFIELDS, req.data) else: raise ValueError( 'HTTP method "{}" not supported in HTTPS-in-HTTPS mode'.format( method ) ) # Location c.setopt(pycurl.URL, req.get_full_url()) # Headers header_str_list = [ "{}: {}".format(name, val) for name, val in req.header_items() ] if len(header_str_list) > 0: c.setopt(pycurl.HTTPHEADER, header_str_list) # Behavior c.setopt(pycurl.FOLLOWLOCATION, True) c.setopt(pycurl.CAINFO, defaults.SSL_CERTS_PATH) if timeout: c.setopt(pycurl.TIMEOUT, timeout) # Proxy if https_proxy: parsed_https_proxy = _parse_https_proxy(https_proxy) https_proxy = ( parsed_https_proxy.geturl() if parsed_https_proxy else None ) c.setopt(pycurl.PROXY, https_proxy) c.setopt(pycurl.PROXYTYPE, 2) # 2 == HTTPS else: LOG.warning("in pycurl request function without an https proxy") # Response handling body_output = io.BytesIO() c.setopt(pycurl.WRITEDATA, body_output) headers = {} def save_header(header_line): header_line = header_line.decode("iso-8859-1") if ":" not in header_line: return name_raw, value_raw = header_line.split(":", 1) name = name_raw.strip().lower() value = value_raw.strip() headers[name] = value c.setopt(pycurl.HEADERFUNCTION, save_header) # Do it try: c.perform() except pycurl.error as e: _handle_pycurl_error( e, url=req.get_full_url(), authentication_error_code=pycurl.E_RECV_ERROR, ca_certificates_error_code=pycurl.E_SSL_CACERT_BADFILE, ) code = int(c.getinfo(pycurl.RESPONSE_CODE)) body = body_output.getvalue().decode("utf-8") c.close() return UnparsedHTTPResponse( code=code, headers=headers, body=body, ) def _parse_https_proxy(https_proxy) -> Optional[ParseResult]: if not https_proxy: https_proxy = request.getproxies().get("https") return urlparse(https_proxy) if https_proxy else None def readurl( url: str, data: Optional[bytes] = None, headers: Dict[str, str] = {}, method: Optional[str] = None, timeout: Optional[int] = None, log_response_body: bool = True, ) -> HTTPResponse: if not is_service_url(url): raise exceptions.InvalidUrl(url=url) if data and not method: method = "POST" req = request.Request(url, data=data, headers=headers, method=method) sorted_header_str = ", ".join( ["'{}': '{}'".format(k, headers[k]) for k in sorted(headers)] ) LOG.debug( "URL [{}]: {}, headers: {{{}}}, data: {}".format( method or "GET", url, sorted_header_str, data.decode("utf-8") if data else None, ) ) https_proxy = get_configured_web_proxy().get("https") if should_use_pycurl(https_proxy, url): resp = _readurl_pycurl_https_in_https( req, timeout=timeout, https_proxy=https_proxy ) else: resp = _readurl_urllib(req, timeout=timeout) json_dict = {} json_list = [] if "application/json" in resp.headers.get("content-type", ""): json_body = json.loads(resp.body, cls=util.DatetimeAwareJSONDecoder) if isinstance(json_body, dict): json_dict = json_body elif isinstance(json_body, list): json_list = json_body sorted_header_str = ", ".join( ["'{}': '{}'".format(k, resp.headers[k]) for k in sorted(resp.headers)] ) debug_msg = "URL [{}] response: {}, headers: {{{}}}".format( method or "GET", url, sorted_header_str ) if log_response_body: # Due to implicit logging redaction, large responses might take longer body_to_log = resp.body # type: Any if json_dict: body_to_log = json_dict elif json_list: body_to_log = json_list debug_msg += ", data: {}".format(body_to_log) LOG.debug(debug_msg) return HTTPResponse( code=resp.code, headers=resp.headers, body=resp.body, json_dict=json_dict, json_list=json_list, )
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings