AnonSec Shell

. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . AnonSec Shell AnonSec Shell

Server IP : 172.67.142.142 / Your IP : 104.23.243.117 [ Reverse IP ]
Web Server : nginx/1.18.0
System : Linux ip-172-31-29-104 5.15.0-1075-aws #82~20.04.1-Ubuntu SMP Thu Dec 19 05:24:09 UTC 2024 x86_64
User : www-data ( 33)
PHP Version : 7.4.3-4ubuntu2.29
Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
Domains : 2 Domains
MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON
Directory : /lib/python3/dist-packages/twisted/

Upload File :

[ HOME ] [ BACKUP SHELL ] [ JUMPING ] [ MASS DEFACE ] [ SCAN ROOT ] [ SYMLINK ]

Current File : /lib/python3/dist-packages/twisted/plugin.py

# -*- test-case-name: twisted.test.test_plugin -*-
# Copyright (c) 2005 Divmod, Inc.
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Plugin system for Twisted.

@author: Jp Calderone
@author: Glyph Lefkowitz
"""

from __future__ import absolute_import, division

import os
import sys

from zope.interface import Interface, providedBy

def _determinePickleModule():
    """
    Determine which 'pickle' API module to use.
    """
    try:
        import cPickle
        return cPickle
    except ImportError:
        import pickle
        return pickle

pickle = _determinePickleModule()

from twisted.python.components import getAdapterFactory
from twisted.python.reflect import namedAny
from twisted.python import log
from twisted.python.modules import getModule
from twisted.python.compat import iteritems



class IPlugin(Interface):
    """
    Interface that must be implemented by all plugins.

    Only objects which implement this interface will be considered for return
    by C{getPlugins}.  To be useful, plugins should also implement some other
    application-specific interface.
    """



class CachedPlugin(object):
    def __init__(self, dropin, name, description, provided):
        self.dropin = dropin
        self.name = name
        self.description = description
        self.provided = provided
        self.dropin.plugins.append(self)

    def __repr__(self):
        return '<CachedPlugin %r/%r (provides %r)>' % (
            self.name, self.dropin.moduleName,
            ', '.join([i.__name__ for i in self.provided]))

    def load(self):
        return namedAny(self.dropin.moduleName + '.' + self.name)

    def __conform__(self, interface, registry=None, default=None):
        for providedInterface in self.provided:
            if providedInterface.isOrExtends(interface):
                return self.load()
            if getAdapterFactory(providedInterface, interface, None) is not None:
                return interface(self.load(), default)
        return default

    # backwards compat HOORJ
    getComponent = __conform__



class CachedDropin(object):
    """
    A collection of L{CachedPlugin} instances from a particular module in a
    plugin package.

    @type moduleName: C{str}
    @ivar moduleName: The fully qualified name of the plugin module this
        represents.

    @type description: C{str} or L{None}
    @ivar description: A brief explanation of this collection of plugins
        (probably the plugin module's docstring).

    @type plugins: C{list}
    @ivar plugins: The L{CachedPlugin} instances which were loaded from this
        dropin.
    """
    def __init__(self, moduleName, description):
        self.moduleName = moduleName
        self.description = description
        self.plugins = []



def _generateCacheEntry(provider):
    dropin = CachedDropin(provider.__name__,
                          provider.__doc__)
    for k, v in iteritems(provider.__dict__):
        plugin = IPlugin(v, None)
        if plugin is not None:
            # Instantiated for its side-effects.
            CachedPlugin(dropin, k, v.__doc__, list(providedBy(plugin)))
    return dropin

try:
    fromkeys = dict.fromkeys
except AttributeError:
    def fromkeys(keys, value=None):
        d = {}
        for k in keys:
            d[k] = value
        return d



def getCache(module):
    """
    Compute all the possible loadable plugins, while loading as few as
    possible and hitting the filesystem as little as possible.

    @param module: a Python module object.  This represents a package to search
    for plugins.

    @return: a dictionary mapping module names to L{CachedDropin} instances.
    """
    allCachesCombined = {}
    mod = getModule(module.__name__)
    # don't want to walk deep, only immediate children.
    buckets = {}
    # Fill buckets with modules by related entry on the given package's
    # __path__.  There's an abstraction inversion going on here, because this
    # information is already represented internally in twisted.python.modules,
    # but it's simple enough that I'm willing to live with it.  If anyone else
    # wants to fix up this iteration so that it's one path segment at a time,
    # be my guest.  --glyph
    for plugmod in mod.iterModules():
        fpp = plugmod.filePath.parent()
        if fpp not in buckets:
            buckets[fpp] = []
        bucket = buckets[fpp]
        bucket.append(plugmod)
    for pseudoPackagePath, bucket in iteritems(buckets):
        dropinPath = pseudoPackagePath.child('dropin.cache')
        try:
            lastCached = dropinPath.getModificationTime()
            with dropinPath.open('r') as f:
                dropinDotCache = pickle.load(f)
        except:
            dropinDotCache = {}
            lastCached = 0

        needsWrite = False
        existingKeys = {}
        for pluginModule in bucket:
            pluginKey = pluginModule.name.split('.')[-1]
            existingKeys[pluginKey] = True
            if ((pluginKey not in dropinDotCache) or
                (pluginModule.filePath.getModificationTime() >= lastCached)):
                needsWrite = True
                try:
                    provider = pluginModule.load()
                except:
                    # dropinDotCache.pop(pluginKey, None)
                    log.err()
                else:
                    entry = _generateCacheEntry(provider)
                    dropinDotCache[pluginKey] = entry
        # Make sure that the cache doesn't contain any stale plugins.
        for pluginKey in list(dropinDotCache.keys()):
            if pluginKey not in existingKeys:
                del dropinDotCache[pluginKey]
                needsWrite = True
        if needsWrite:
            try:
                dropinPath.setContent(pickle.dumps(dropinDotCache))
            except OSError as e:
                log.msg(
                    format=(
                        "Unable to write to plugin cache %(path)s: error "
                        "number %(errno)d"),
                    path=dropinPath.path, errno=e.errno)
            except:
                log.err(None, "Unexpected error while writing cache file")
        allCachesCombined.update(dropinDotCache)
    return allCachesCombined



def getPlugins(interface, package=None):
    """
    Retrieve all plugins implementing the given interface beneath the given module.

    @param interface: An interface class.  Only plugins which implement this
    interface will be returned.

    @param package: A package beneath which plugins are installed.  For
    most uses, the default value is correct.

    @return: An iterator of plugins.
    """
    if package is None:
        import twisted.plugins as package
    allDropins = getCache(package)
    for key, dropin in iteritems(allDropins):
        for plugin in dropin.plugins:
            try:
                adapted = interface(plugin, None)
            except:
                log.err()
            else:
                if adapted is not None:
                    yield adapted


# Old, backwards compatible name.  Don't use this.
getPlugIns = getPlugins


def pluginPackagePaths(name):
    """
    Return a list of additional directories which should be searched for
    modules to be included as part of the named plugin package.

    @type name: C{str}
    @param name: The fully-qualified Python name of a plugin package, eg
        C{'twisted.plugins'}.

    @rtype: C{list} of C{str}
    @return: The absolute paths to other directories which may contain plugin
        modules for the named plugin package.
    """
    package = name.split('.')
    # Note that this may include directories which do not exist.  It may be
    # preferable to remove such directories at this point, rather than allow
    # them to be searched later on.
    #
    # Note as well that only '__init__.py' will be considered to make a
    # directory a package (and thus exclude it from this list).  This means
    # that if you create a master plugin package which has some other kind of
    # __init__ (eg, __init__.pyc) it will be incorrectly treated as a
    # supplementary plugin directory.
    return [
        os.path.abspath(os.path.join(x, *package))
        for x
        in sys.path
        if
        not os.path.exists(os.path.join(x, *package + ['__init__.py']))]

__all__ = ['getPlugins', 'pluginPackagePaths']

Anon7 - 2022
AnonSec Team