?HAMA

Edit File Size: 2.23 KB

/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/model/tls_check.py

Text Base64

import logging
import threading
import time
import traceback

from playhouse.sqlite_ext import SqliteExtDatabase

from defence360agent.internals.global_scope import g

class OverridingReset(Exception):
    """
    Overriding reset could be a signal of logic error
    thus need to be explicitly handled in all places where
    this exception is expected to occur.
    """

pass

logger = logging.getLogger(__name__)
_thread_local_storage = threading.local()

_SLOW_TXN_THRESHOLD_S = 5.0

class _TimedAtomic:
    def __init__(self, inner: object):
        self._inner = inner
        self._start: float = 0.0
        self._caller: str = ""

def __enter__(self):
        self._start = time.monotonic()
        self._caller = "".join(traceback.format_stack(limit=4)[:-1])
        return self._inner.__enter__()

def __exit__(self, *args):
        result = self._inner.__exit__(*args)
        elapsed = time.monotonic() - self._start
        if elapsed > _SLOW_TXN_THRESHOLD_S:
            logger.warning(
                "Slow transaction held for %.2fs\n%s",
                elapsed,
                self._caller,
            )
        return result

class SqliteDatabaseWrapper(SqliteExtDatabase):
    def execute_sql(self, *args, **kwargs):
        _validate(*args, **kwargs)
        return super().execute_sql(*args, **kwargs)

def atomic(self, lock_type: str = "IMMEDIATE"):
        inner = super().atomic(lock_type)
        if g.get("DEBUG"):
            return _TimedAtomic(inner)
        return inner

def reset(new_value=None):
    if hasattr(_thread_local_storage, "thread_ident_memo"):
        raise OverridingReset()

_thread_local_storage.thread_ident_memo = (
        new_value or threading.get_ident()
    )

def _validate(*args, **kwargs):
    thread_ident_memo = getattr(
        _thread_local_storage, "thread_ident_memo", None
    )

if thread_ident_memo is None:
        logger.error("wrong thread or _validate() was not preceded by reset()")

elif thread_ident_memo != threading.get_ident():
        logger.error(
            "thread_ident_memo check failed [%r != %r]\n"
            "context:\nargs: %s\nkwargs: %s",
            thread_ident_memo,
            threading.get_ident(),
            args,
            kwargs,
        )

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2026-06-08 20:24:30	Edit Download recursive recursive
analyst_cleanup.py	3.40 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
event_hook.py	1.74 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
icontact.py	1.16 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
infected_domain.py	4.24 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
instance.py	537 B	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
messages_to_send.py	1.37 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
simplification.py	5.01 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
tls_check.py	2.23 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
wordpress.py	1.66 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
wordpress_incident.py	13.02 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
wp_disabled_rule.py	13.21 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
__init__.py	963 B	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive

Directory Contents