ACIL FM

Nama	Ukuran	Aksi
django	-	Open
spark	-	Open
__pycache__	-	Open
aiohttp.py	7.77 MB	View DL Edit
argv.py	945 B	View DL Edit
asgi.py	8.17 MB	View DL Edit
atexit.py	1.79 MB	View DL Edit
aws_lambda.py	12.64 MB	View DL Edit
beam.py	5.53 MB	View DL Edit
boto3.py	3.64 MB	View DL Edit
bottle.py	6.04 MB	View DL Edit
celery.py	8.91 MB	View DL Edit
chalice.py	4.47 MB	View DL Edit
dedupe.py	1.14 MB	View DL Edit
excepthook.py	2.14 MB	View DL Edit
executing.py	1.98 MB	View DL Edit
falcon.py	6.64 MB	View DL Edit
flask.py	7.25 MB	View DL Edit
gcp.py	7.21 MB	View DL Edit
gnu_backtrace.py	2.84 MB	View DL Edit
logging.py	7.4 MB	View DL Edit
modules.py	1.36 MB	View DL Edit
pure_eval.py	4.41 MB	View DL Edit
pyramid.py	6.91 MB	View DL Edit
redis.py	3 MB	View DL Edit
rq.py	4.68 MB	View DL Edit
sanic.py	7.53 MB	View DL Edit
serverless.py	1.92 MB	View DL Edit
sqlalchemy.py	2.87 MB	View DL Edit
stdlib.py	7.18 MB	View DL Edit
threading.py	2.79 MB	View DL Edit
tornado.py	6.84 MB	View DL Edit
trytond.py	1.69 MB	View DL Edit
wsgi.py	10.13 MB	View DL Edit
_wsgi_common.py	4.65 MB	View DL Edit
__init__.py	6.36 MB	View DL Edit

Edit file: /opt/imunify360/venv/lib/python3.11/site-packages/sentry_sdk/integrations/asgi.py

"""
An ASGI middleware.

Based on Tom Christie's `sentry-asgi <https://github.com/encode/sentry-asgi>`_.
"""

import asyncio
import inspect
import urllib

from sentry_sdk._functools import partial
from sentry_sdk._types import MYPY
from sentry_sdk.hub import Hub, _should_send_default_pii
from sentry_sdk.integrations._wsgi_common import _filter_headers
from sentry_sdk.utils import (
    ContextVar,
    event_from_exception,
    transaction_from_function,
    HAS_REAL_CONTEXTVARS,
    CONTEXTVARS_ERROR_MESSAGE,
)
from sentry_sdk.tracing import Transaction

if MYPY:
    from typing import Dict
    from typing import Any
    from typing import Optional
    from typing import Callable

from typing_extensions import Literal

from sentry_sdk._types import Event, Hint

_asgi_middleware_applied = ContextVar("sentry_asgi_middleware_applied")

_DEFAULT_TRANSACTION_NAME = "generic ASGI request"

def _capture_exception(hub, exc):
    # type: (Hub, Any) -> None

# Check client here as it might have been unset while streaming response
    if hub.client is not None:
        event, hint = event_from_exception(
            exc,
            client_options=hub.client.options,
            mechanism={"type": "asgi", "handled": False},
        )
        hub.capture_event(event, hint=hint)

def _looks_like_asgi3(app):
    # type: (Any) -> bool
    """
    Try to figure out if an application object supports ASGI3.

This is how uvicorn figures out the application version as well.
    """
    if inspect.isclass(app):
        return hasattr(app, "__await__")
    elif inspect.isfunction(app):
        return asyncio.iscoroutinefunction(app)
    else:
        call = getattr(app, "__call__", None)  # noqa
        return asyncio.iscoroutinefunction(call)

class SentryAsgiMiddleware:
    __slots__ = ("app", "__call__")

def __init__(self, app, unsafe_context_data=False):
        # type: (Any, bool) -> None
        """
        Instrument an ASGI application with Sentry. Provides HTTP/websocket
        data to sent events and basic handling for exceptions bubbling up
        through the middleware.

:param unsafe_context_data: Disable errors when a proper contextvars installation could not be found. We do not recommend changing this from the default.
        """

if not unsafe_context_data and not HAS_REAL_CONTEXTVARS:
            # We better have contextvars or we're going to leak state between
            # requests.
            raise RuntimeError(
                "The ASGI middleware for Sentry requires Python 3.7+ "
                "or the aiocontextvars package." + CONTEXTVARS_ERROR_MESSAGE
            )
        self.app = app

if _looks_like_asgi3(app):
            self.__call__ = self._run_asgi3  # type: Callable[..., Any]
        else:
            self.__call__ = self._run_asgi2

def _run_asgi2(self, scope):
        # type: (Any) -> Any
        async def inner(receive, send):
            # type: (Any, Any) -> Any
            return await self._run_app(scope, lambda: self.app(scope)(receive, send))

return inner

async def _run_asgi3(self, scope, receive, send):
        # type: (Any, Any, Any) -> Any
        return await self._run_app(scope, lambda: self.app(scope, receive, send))

async def _run_app(self, scope, callback):
        # type: (Any, Any) -> Any
        is_recursive_asgi_middleware = _asgi_middleware_applied.get(False)

if is_recursive_asgi_middleware:
            try:
                return await callback()
            except Exception as exc:
                _capture_exception(Hub.current, exc)
                raise exc from None

_asgi_middleware_applied.set(True)
        try:
            hub = Hub(Hub.current)
            with hub:
                with hub.configure_scope() as sentry_scope:
                    sentry_scope.clear_breadcrumbs()
                    sentry_scope._name = "asgi"
                    processor = partial(self.event_processor, asgi_scope=scope)
                    sentry_scope.add_event_processor(processor)

ty = scope["type"]

if ty in ("http", "websocket"):
                    transaction = Transaction.continue_from_headers(
                        dict(scope["headers"]),
                        op="{}.server".format(ty),
                    )
                else:
                    transaction = Transaction(op="asgi.server")

transaction.name = _DEFAULT_TRANSACTION_NAME
                transaction.set_tag("asgi.type", ty)

with hub.start_transaction(transaction):
                    # XXX: Would be cool to have correct span status, but we
                    # would have to wrap send(). That is a bit hard to do with
                    # the current abstraction over ASGI 2/3.
                    try:
                        return await callback()
                    except Exception as exc:
                        _capture_exception(hub, exc)
                        raise exc from None
        finally:
            _asgi_middleware_applied.set(False)

def event_processor(self, event, hint, asgi_scope):
        # type: (Event, Hint, Any) -> Optional[Event]
        request_info = event.get("request", {})

ty = asgi_scope["type"]
        if ty in ("http", "websocket"):
            request_info["method"] = asgi_scope.get("method")
            request_info["headers"] = headers = _filter_headers(
                self._get_headers(asgi_scope)
            )
            request_info["query_string"] = self._get_query(asgi_scope)

request_info["url"] = self._get_url(
                asgi_scope, "http" if ty == "http" else "ws", headers.get("host")
            )

client = asgi_scope.get("client")
        if client and _should_send_default_pii():
            request_info["env"] = {"REMOTE_ADDR": client[0]}

if (
            event.get("transaction", _DEFAULT_TRANSACTION_NAME)
            == _DEFAULT_TRANSACTION_NAME
        ):
            endpoint = asgi_scope.get("endpoint")
            # Webframeworks like Starlette mutate the ASGI env once routing is
            # done, which is sometime after the request has started. If we have
            # an endpoint, overwrite our generic transaction name.
            if endpoint:
                event["transaction"] = transaction_from_function(endpoint)

event["request"] = request_info

return event

# Helper functions for extracting request data.
    #
    # Note: Those functions are not public API. If you want to mutate request
    # data to your liking it's recommended to use the `before_send` callback
    # for that.

def _get_url(self, scope, default_scheme, host):
        # type: (Dict[str, Any], Literal["ws", "http"], Optional[str]) -> str
        """
        Extract URL from the ASGI scope, without also including the querystring.
        """
        scheme = scope.get("scheme", default_scheme)

server = scope.get("server", None)
        path = scope.get("root_path", "") + scope.get("path", "")

if host:
            return "%s://%s%s" % (scheme, host, path)

if server is not None:
            host, port = server
            default_port = {"http": 80, "https": 443, "ws": 80, "wss": 443}[scheme]
            if port != default_port:
                return "%s://%s:%s%s" % (scheme, host, port, path)
            return "%s://%s%s" % (scheme, host, path)
        return path

def _get_query(self, scope):
        # type: (Any) -> Any
        """
        Extract querystring from the ASGI scope, in the format that the Sentry protocol expects.
        """
        qs = scope.get("query_string")
        if not qs:
            return None
        return urllib.parse.unquote(qs.decode("latin-1"))

def _get_headers(self, scope):
        # type: (Any) -> Dict[str, str]
        """
        Extract headers from the ASGI scope, in the format that the Sentry protocol expects.
        """
        headers = {}  # type: Dict[str, str]
        for raw_key, raw_value in scope["headers"]:
            key = raw_key.decode("latin-1")
            value = raw_value.decode("latin-1")
            if key in headers:
                headers[key] = headers[key] + ", " + value
            else:
                headers[key] = value
        return headers

Batal