FM / botocore

/opt/imh-python/lib/python3.9/site-packages/botocore/

/opt/imh-python/lib/python3.9/site-packages/botocore

Name	Size	Mode	Actions
data/	-	0755	rm
docs/	-	0755	rm
retries/	-	0755	rm
vendored/	-	0755	rm
__pycache__/	-	0755	rm
args.py	16199	0644	edit dl rm
auth.py	34966	0644	edit dl rm
awsrequest.py	23247	0644	edit dl rm
cacert.pem	271088	0644	edit dl rm
client.py	41514	0644	edit dl rm
compat.py	11715	0644	edit dl rm
config.py	12721	0644	edit dl rm
configloader.py	9580	0644	edit dl rm
configprovider.py	22235	0644	edit dl rm
credentials.py	81582	0644	edit dl rm
discovery.py	11031	0644	edit dl rm
endpoint.py	13811	0644	edit dl rm
errorfactory.py	3727	0644	edit dl rm
eventstream.py	20517	0644	edit dl rm
exceptions.py	19927	0644	edit dl rm
handlers.py	45142	0644	edit dl rm
history.py	1748	0644	edit dl rm
hooks.py	24573	0644	edit dl rm
httpsession.py	13547	0644	edit dl rm
loaders.py	17355	0644	edit dl rm
model.py	28352	0644	edit dl rm
monitoring.py	20586	0644	edit dl rm
paginate.py	27128	0644	edit dl rm
parsers.py	42301	0644	edit dl rm
regions.py	8975	0644	edit dl rm
response.py	6434	0644	edit dl rm
retryhandler.py	13781	0644	edit dl rm
serialize.py	30430	0644	edit dl rm
session.py	43433	0644	edit dl rm
signers.py	28234	0644	edit dl rm
stub.py	14361	0644	edit dl rm
translate.py	3412	0644	edit dl rm
utils.py	86492	0644	edit dl rm
validate.py	11323	0644	edit dl rm
waiter.py	13860	0644	edit dl rm
__init__.py	3881	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/botocore/monitoring.py (20586B)

# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
#     http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file 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 json
import logging
import re
import time

from botocore.compat import ensure_unicode, ensure_bytes, urlparse
from botocore.retryhandler import EXCEPTION_MAP as RETRYABLE_EXCEPTIONS

logger = logging.getLogger(__name__)

class Monitor(object):
    _EVENTS_TO_REGISTER = [
        'before-parameter-build',
        'request-created',
        'response-received',
        'after-call',
        'after-call-error',
    ]

def __init__(self, adapter, publisher):
        """Abstraction for monitoring clients API calls

:param adapter: An adapter that takes event emitter events
            and produces monitor events

:param publisher: A publisher for generated monitor events
        """
        self._adapter = adapter
        self._publisher = publisher

def register(self, event_emitter):
        """Register an event emitter to the monitor"""
        for event_to_register in self._EVENTS_TO_REGISTER:
            event_emitter.register_last(event_to_register, self.capture)

def capture(self, event_name, **payload):
        """Captures an incoming event from the event emitter

It will feed an event emitter event to the monitor's adaptor to create
        a monitor event and then publish that event to the monitor's publisher.
        """
        try:
            monitor_event = self._adapter.feed(event_name, payload)
            if monitor_event:
                self._publisher.publish(monitor_event)
        except Exception as e:
            logger.debug(
                'Exception %s raised by client monitor in handling event %s',
                e, event_name, exc_info=True)

class MonitorEventAdapter(object):
    def __init__(self, time=time.time):
        """Adapts event emitter events to produce monitor events

:type time: callable
        :param time: A callable that produces the current time
        """
        self._time = time

def feed(self, emitter_event_name, emitter_payload):
        """Feed an event emitter event to generate a monitor event

:type emitter_event_name: str
        :param emitter_event_name: The name of the event emitted

:type emitter_payload: dict
        :param emitter_payload: The payload to associated to the event
            emitted

:rtype: BaseMonitorEvent
        :returns: A monitor event based on the event emitter events
            fired
        """
        return self._get_handler(emitter_event_name)(**emitter_payload)

def _get_handler(self, event_name):
        return getattr(
            self, '_handle_' + event_name.split('.')[0].replace('-', '_')
        )

def _handle_before_parameter_build(self, model, context, **kwargs):
        context['current_api_call_event'] = APICallEvent(
            service=model.service_model.service_id,
            operation=model.wire_name,
            timestamp=self._get_current_time(),
        )

def _handle_request_created(self, request, **kwargs):
        context = request.context
        new_attempt_event = context[
            'current_api_call_event'].new_api_call_attempt(
                timestamp=self._get_current_time())
        new_attempt_event.request_headers = request.headers
        new_attempt_event.url = request.url
        context['current_api_call_attempt_event'] = new_attempt_event

def _handle_response_received(self, parsed_response, context, exception,
                                  **kwargs):
        attempt_event = context.pop('current_api_call_attempt_event')
        attempt_event.latency = self._get_latency(attempt_event)
        if parsed_response is not None:
            attempt_event.http_status_code = parsed_response[
                'ResponseMetadata']['HTTPStatusCode']
            attempt_event.response_headers = parsed_response[
                'ResponseMetadata']['HTTPHeaders']
            attempt_event.parsed_error = parsed_response.get('Error')
        else:
            attempt_event.wire_exception = exception
        return attempt_event

def _handle_after_call(self, context, parsed, **kwargs):
        context['current_api_call_event'].retries_exceeded = parsed[
            'ResponseMetadata'].get('MaxAttemptsReached', False)
        return self._complete_api_call(context)

def _handle_after_call_error(self, context, exception, **kwargs):
        # If the after-call-error was emitted and the error being raised
        # was a retryable connection error, then the retries must have exceeded
        # for that exception as this event gets emitted **after** retries
        # happen.
        context['current_api_call_event'].retries_exceeded = \
            self._is_retryable_exception(exception)
        return self._complete_api_call(context)

def _is_retryable_exception(self, exception):
        return isinstance(
            exception, tuple(RETRYABLE_EXCEPTIONS['GENERAL_CONNECTION_ERROR']))

def _complete_api_call(self, context):
        call_event = context.pop('current_api_call_event')
        call_event.latency = self._get_latency(call_event)
        return call_event

def _get_latency(self, event):
        return self._get_current_time() - event.timestamp

def _get_current_time(self):
        return int(self._time() * 1000)

class BaseMonitorEvent(object):
    def __init__(self, service, operation, timestamp):
        """Base monitor event

:type service: str
        :param service: A string identifying the service associated to
            the event

:type operation: str
        :param operation: A string identifying the operation of service
            associated to the event

:type timestamp: int
        :param timestamp: Epoch time in milliseconds from when the event began
        """
        self.service = service
        self.operation = operation
        self.timestamp = timestamp

def __repr__(self):
        return '%s(%r)' % (self.__class__.__name__, self.__dict__)

def __eq__(self, other):
        if isinstance(other, self.__class__):
            return self.__dict__ == other.__dict__
        return False

class APICallEvent(BaseMonitorEvent):
    def __init__(self, service, operation, timestamp, latency=None,
                 attempts=None, retries_exceeded=False):
        """Monitor event for a single API call

This event corresponds to a single client method call, which includes
        every HTTP requests attempt made in order to complete the client call

:type service: str
        :param service: A string identifying the service associated to
            the event

:type operation: str
        :param operation: A string identifying the operation of service
            associated to the event

:type timestamp: int
        :param timestamp: Epoch time in milliseconds from when the event began

:type latency: int
        :param latency: The time in milliseconds to complete the client call

:type attempts: list
        :param attempts: The list of APICallAttempts associated to the
            APICall

:type retries_exceeded: bool
        :param retries_exceeded: True if API call exceeded retries. False
            otherwise
        """
        super(APICallEvent, self).__init__(
            service=service, operation=operation, timestamp=timestamp)
        self.latency = latency
        self.attempts = attempts
        if attempts is None:
            self.attempts = []
        self.retries_exceeded = retries_exceeded

def new_api_call_attempt(self, timestamp):
        """Instantiates APICallAttemptEvent associated to the APICallEvent

:type timestamp: int
        :param timestamp: Epoch time in milliseconds to associate to the
            APICallAttemptEvent
        """
        attempt_event = APICallAttemptEvent(
            service=self.service,
            operation=self.operation,
            timestamp=timestamp
        )
        self.attempts.append(attempt_event)
        return attempt_event

class APICallAttemptEvent(BaseMonitorEvent):
    def __init__(self, service, operation, timestamp,
                 latency=None, url=None, http_status_code=None,
                 request_headers=None, response_headers=None,
                 parsed_error=None, wire_exception=None):
        """Monitor event for a single API call attempt

This event corresponds to a single HTTP request attempt in completing
        the entire client method call.

:type service: str
        :param service: A string identifying the service associated to
            the event

:type operation: str
        :param operation: A string identifying the operation of service
            associated to the event

:type timestamp: int
        :param timestamp: Epoch time in milliseconds from when the HTTP request
            started

:type latency: int
        :param latency: The time in milliseconds to complete the HTTP request
            whether it succeeded or failed

:type url: str
        :param url: The URL the attempt was sent to

:type http_status_code: int
        :param http_status_code: The HTTP status code of the HTTP response
            if there was a response

:type request_headers: dict
        :param request_headers: The HTTP headers sent in making the HTTP
            request

:type response_headers: dict
        :param response_headers: The HTTP headers returned in the HTTP response
            if there was a response

:type parsed_error: dict
        :param parsed_error: The error parsed if the service returned an
            error back

:type wire_exception: Exception
        :param wire_exception: The exception raised in sending the HTTP
            request (i.e. ConnectionError)
        """
        super(APICallAttemptEvent, self).__init__(
            service=service, operation=operation, timestamp=timestamp
        )
        self.latency = latency
        self.url = url
        self.http_status_code = http_status_code
        self.request_headers = request_headers
        self.response_headers = response_headers
        self.parsed_error = parsed_error
        self.wire_exception = wire_exception

class CSMSerializer(object):
    _MAX_CLIENT_ID_LENGTH = 255
    _MAX_EXCEPTION_CLASS_LENGTH = 128
    _MAX_ERROR_CODE_LENGTH = 128
    _MAX_USER_AGENT_LENGTH = 256
    _MAX_MESSAGE_LENGTH = 512
    _RESPONSE_HEADERS_TO_EVENT_ENTRIES = {
        'x-amzn-requestid': 'XAmznRequestId',
        'x-amz-request-id': 'XAmzRequestId',
        'x-amz-id-2': 'XAmzId2',
    }
    _AUTH_REGEXS = {
        'v4': re.compile(
            r'AWS4-HMAC-SHA256 '
            r'Credential=(?P<access_key>\w+)/\d+/'
            r'(?P<signing_region>[a-z0-9-]+)/'
        ),
        's3': re.compile(
            r'AWS (?P<access_key>\w+):'
        )
    }
    _SERIALIZEABLE_EVENT_PROPERTIES = [
        'service',
        'operation',
        'timestamp',
        'attempts',
        'latency',
        'retries_exceeded',
        'url',
        'request_headers',
        'http_status_code',
        'response_headers',
        'parsed_error',
        'wire_exception',
    ]

def __init__(self, csm_client_id):
        """Serializes monitor events to CSM (Client Side Monitoring) format

:type csm_client_id: str
        :param csm_client_id: The application identifier to associate
            to the serialized events
        """
        self._validate_client_id(csm_client_id)
        self.csm_client_id = csm_client_id

def _validate_client_id(self, csm_client_id):
        if len(csm_client_id) > self._MAX_CLIENT_ID_LENGTH:
            raise ValueError(
                'The value provided for csm_client_id: %s exceeds the '
                'maximum length of %s characters' % (
                    csm_client_id, self._MAX_CLIENT_ID_LENGTH)
            )

def serialize(self, event):
        """Serializes a monitor event to the CSM format

:type event: BaseMonitorEvent
        :param event: The event to serialize to bytes

:rtype: bytes
        :returns: The CSM serialized form of the event
        """
        event_dict = self._get_base_event_dict(event)
        event_type = self._get_event_type(event)
        event_dict['Type'] = event_type
        for attr in self._SERIALIZEABLE_EVENT_PROPERTIES:
            value = getattr(event, attr, None)
            if value is not None:
                getattr(self, '_serialize_' + attr)(
                    value, event_dict, event_type=event_type)
        return ensure_bytes(
            json.dumps(event_dict, separators=(',', ':')))

def _get_base_event_dict(self, event):
        return {
            'Version': 1,
            'ClientId': self.csm_client_id,
        }

def _serialize_service(self, service, event_dict, **kwargs):
        event_dict['Service'] = service

def _serialize_operation(self, operation, event_dict, **kwargs):
        event_dict['Api'] = operation

def _serialize_timestamp(self, timestamp, event_dict, **kwargs):
        event_dict['Timestamp'] = timestamp

def _serialize_attempts(self, attempts, event_dict, **kwargs):
        event_dict['AttemptCount'] = len(attempts)
        if attempts:
            self._add_fields_from_last_attempt(event_dict, attempts[-1])

def _add_fields_from_last_attempt(self, event_dict, last_attempt):
        if last_attempt.request_headers:
            # It does not matter which attempt to use to grab the region
            # for the ApiCall event, but SDKs typically do the last one.
            region = self._get_region(last_attempt.request_headers)
            if region is not None:
                event_dict['Region'] = region
            event_dict['UserAgent'] = self._get_user_agent(
                last_attempt.request_headers)
        if last_attempt.http_status_code is not None:
            event_dict['FinalHttpStatusCode'] = last_attempt.http_status_code
        if last_attempt.parsed_error is not None:
            self._serialize_parsed_error(
                last_attempt.parsed_error, event_dict, 'ApiCall')
        if last_attempt.wire_exception is not None:
            self._serialize_wire_exception(
                last_attempt.wire_exception, event_dict, 'ApiCall')

def _serialize_latency(self, latency, event_dict, event_type):
        if event_type == 'ApiCall':
            event_dict['Latency'] = latency
        elif event_type == 'ApiCallAttempt':
            event_dict['AttemptLatency'] = latency

def _serialize_retries_exceeded(self, retries_exceeded, event_dict,
                                    **kwargs):
        event_dict['MaxRetriesExceeded'] = (1 if retries_exceeded else 0)

def _serialize_url(self, url, event_dict, **kwargs):
        event_dict['Fqdn'] = urlparse(url).netloc

def _serialize_request_headers(self, request_headers, event_dict,
                                   **kwargs):
        event_dict['UserAgent'] = self._get_user_agent(request_headers)
        if self._is_signed(request_headers):
            event_dict['AccessKey'] = self._get_access_key(request_headers)
        region = self._get_region(request_headers)
        if region is not None:
            event_dict['Region'] = region
        if 'X-Amz-Security-Token' in request_headers:
            event_dict['SessionToken'] = request_headers[
                'X-Amz-Security-Token']

def _serialize_http_status_code(self, http_status_code, event_dict,
                                    **kwargs):
        event_dict['HttpStatusCode'] = http_status_code

def _serialize_response_headers(self, response_headers, event_dict,
                                    **kwargs):
        for header, entry in self._RESPONSE_HEADERS_TO_EVENT_ENTRIES.items():
            if header in response_headers:
                event_dict[entry] = response_headers[header]

def _serialize_parsed_error(self, parsed_error, event_dict, event_type,
                                **kwargs):
        field_prefix = 'Final' if event_type == 'ApiCall' else ''
        event_dict[field_prefix + 'AwsException'] = self._truncate(
            parsed_error['Code'], self._MAX_ERROR_CODE_LENGTH)
        event_dict[field_prefix + 'AwsExceptionMessage'] = self._truncate(
            parsed_error['Message'], self._MAX_MESSAGE_LENGTH)

def _serialize_wire_exception(self, wire_exception, event_dict, event_type,
                                  **kwargs):
        field_prefix = 'Final' if event_type == 'ApiCall' else ''
        event_dict[field_prefix + 'SdkException'] = self._truncate(
            wire_exception.__class__.__name__,
            self._MAX_EXCEPTION_CLASS_LENGTH)
        event_dict[field_prefix + 'SdkExceptionMessage'] = self._truncate(
            str(wire_exception), self._MAX_MESSAGE_LENGTH)

def _get_event_type(self, event):
        if isinstance(event, APICallEvent):
            return 'ApiCall'
        elif isinstance(event, APICallAttemptEvent):
            return 'ApiCallAttempt'

def _get_access_key(self, request_headers):
        auth_val = self._get_auth_value(request_headers)
        _, auth_match = self._get_auth_match(auth_val)
        return auth_match.group('access_key')

def _get_region(self, request_headers):
        if not self._is_signed(request_headers):
            return None
        auth_val = self._get_auth_value(request_headers)
        signature_version, auth_match = self._get_auth_match(auth_val)
        if signature_version != 'v4':
            return None
        return auth_match.group('signing_region')

def _get_user_agent(self, request_headers):
        return self._truncate(
            ensure_unicode(request_headers.get('User-Agent', '')),
            self._MAX_USER_AGENT_LENGTH
        )

def _is_signed(self, request_headers):
        return 'Authorization' in request_headers

def _get_auth_value(self, request_headers):
        return ensure_unicode(request_headers['Authorization'])

def _get_auth_match(self, auth_val):
        for signature_version, regex in self._AUTH_REGEXS.items():
            match = regex.match(auth_val)
            if match:
                return signature_version, match
        return None, None

def _truncate(self, text, max_length):
        if len(text) > max_length:
            logger.debug(
                'Truncating following value to maximum length of '
                '%s: %s', text, max_length)
            return text[:max_length]
        return text

class SocketPublisher(object):
    _MAX_MONITOR_EVENT_LENGTH = 8 * 1024

def __init__(self, socket, host, port, serializer):
        """Publishes monitor events to a socket

:type socket: socket.socket
        :param socket: The socket object to use to publish events

:type host: string
        :param host: The host to send events to

:type port: integer
        :param port: The port on the host to send events to

:param serializer: The serializer to use to serialize the event
            to a form that can be published to the socket. This must
            have a `serialize()` method that accepts a monitor event
            and return bytes
        """
        self._socket = socket
        self._address = (host, port)
        self._serializer = serializer

def publish(self, event):
        """Publishes a specified monitor event

:type event: BaseMonitorEvent
        :param event: The monitor event to be sent
            over the publisher's socket to the desired address.
        """
        serialized_event = self._serializer.serialize(event)
        if len(serialized_event) > self._MAX_MONITOR_EVENT_LENGTH:
            logger.debug(
                'Serialized event of size %s exceeds the maximum length '
                'allowed: %s. Not sending event to socket.',
                len(serialized_event), self._MAX_MONITOR_EVENT_LENGTH
            )
            return
        self._socket.sendto(serialized_event, self._address)

cmd: