ACIL FM

Pilih semua

Nama	Ukuran	Aksi
__pycache__	-	Open
analyst_cleanup.py	3.4 MB	View DL Edit
event_hook.py	1.74 MB	View DL Edit
icontact.py	1.16 MB	View DL Edit
infected_domain.py	4.24 MB	View DL Edit
instance.py	536 B	View DL Edit
messages_to_send.py	1.37 MB	View DL Edit
simplification.py	5.01 MB	View DL Edit
tls_check.py	1.29 MB	View DL Edit
wordpress_incident.py	13.02 MB	View DL Edit
__init__.py	963 B	View DL Edit

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

import logging
import threading

from playhouse.sqlite_ext import SqliteExtDatabase

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()

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

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,
        )

Batal

Isi Zip:

Create