FM / clselectpythonuser

/opt/cloudlinux/venv/lib/python3.11/site-packages/clselect/clselectpythonuser/

/opt/cloudlinux/venv/lib/python3.11/site-packages/clselect/clselectpythonuser

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
environments.py	24548	0644	edit dl rm
extensions.py	6382	0644	edit dl rm
interpreters.py	8126	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/clselect/clselectpythonuser/extensions.py (6382B)

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
import os
import re
from future.moves.http import client as httplib
import socket
from future.moves.xmlrpc import client as xmlrpclib
from future.moves import configparser as ConfigParser
from future.moves.configparser import SafeConfigParser

from pyexpat import ExpatError

import ssl

EXTENSION_PATTERN = re.compile('^(?P<name>.+) $(?P<version>.+)$$',
                               re.MULTILINE)

class PyPIRpcApiError(Exception):
    """
    Generic error class that means that we cannot properly
    get information from xml rpc api
    """
    def __init__(self, message):
        super(PyPIRpcApiError, self).__init__(
            "We are having issues with PyPI RPC api. "
            "You can check https://status.python.org/ to see of "
            "there are any problems with PyPI or contact CloudLinux "
            "support if you see this message for a long time."
            "Original message was: '%s'" % message)

class PyPIMaintenanceException(PyPIRpcApiError):
    """
    Raised when PyPY returns 503 error, which means
    that service is unavailable temporary and we should try again later.
    """
    def __init__(self):
        super(PyPIMaintenanceException, self).__init__(
            "Looks like PyPI is down for maintenance and we are not "
            "able to use it. "
            "You can check https://status.python.org/ to see of "
            "there are any problems with PyPI or contact CloudLinux "
            "support if you see this message for a long time."
        )

class ExtensionInfo(object):

url = 'https://pypi.python.org/pypi'

def __init__(self):
        self._rpc = xmlrpclib.ServerProxy(self.url)

def __get_rpc(self, method, *args, **kwargs):
        try:
            return self._rpc.__getattr__(method)(*args, **kwargs)
        except (socket.gaierror, ExpatError) as err:
            raise PyPIRpcApiError(str(err))
        except xmlrpclib.ProtocolError as err:
            if err.errcode == httplib.SERVICE_UNAVAILABLE:
                raise PyPIMaintenanceException()
            raise PyPIRpcApiError(str(err))
        except xmlrpclib.Fault as err:
            raise PyPIRpcApiError(err.faultString)
        except ssl.SSLError as err:
            raise PyPIRpcApiError(err.strerror)

def list_extensions(self):
        extensions = self.__get_rpc("list_packages")
        return ExtensionInfo.extensions_docs(extensions)

def list_extensions_version(self, extensions):
        # Parameter "True" means show old versions too
        # https://wiki.python.org/moin/PyPIXmlRpc   ( `show_hidden` )
        return dict((extension, {'versions': self.__get_rpc("package_releases", extension, True)})
                    for extension in extensions)

@staticmethod
    def extension_doc(extension):
        return str.join('/', (ExtensionInfo.url, extension))

@staticmethod
    def extensions_docs(extensions):
        docs = (ExtensionInfo.extension_doc(extension)
                for extension in extensions)
        return dict((extension, {'doc': doc})
                    for extension, doc in zip(extensions, docs))

@staticmethod
    def _canonicalise_extension_name(name):
        """
        Normalise a PyPI package name for the locked-extensions lookup.

Per PEP 503, package names are case-insensitive and `-`/`_`/`.` are
        equivalent — `Requests`, `requests`, `re_quests`, `re-quests` all
        refer to the same distribution. Strip whitespace, lower-case, and
        collapse separators so the dict-key check below cannot be bypassed
        by a benign-looking variant of a locked package.
        """
        if not isinstance(name, str):
            return name
        normalised = name.strip().lower()
        normalised = normalised.replace('_', '-').replace('.', '-')
        return normalised

# scanner-triage: interpreter cannot be attacker-shaped. Every caller builds it as
    # a literal-prefixed string ('python'+name, 'ruby'+name, 'node'+name) or via
    # os.path.basename (strips separators); the CLI --version flag is dict-validated
    # against installed interpreters, so no `/`-prefixed value reaches os.path.join.
    @staticmethod
    def get_locked_extensions(interpreter):
        alt_ver = interpreter.replace('.','')
        file_path = os.path.join('/opt/alt', alt_ver ,'etc', 'locked_extensions.ini')
        if not os.path.exists(file_path):
            file_path = os.path.join(os.path.dirname(__file__), '..', 'locked_extensions.ini')
        parser = SafeConfigParser(interpolation=None, strict=False)
        parser.read(file_path)
        try:
            items = parser.items(interpreter)
        except ConfigParser.NoSectionError:
            items = ()

# Canonicalise keys on load so lookups are normalisation-symmetric:
        # whether the caller passes `Requests` or `re-quests`, the membership
        # test against the locked dict still matches.
        return dict(
            (ExtensionInfo._canonicalise_extension_name(extension),
             [v.strip() for v in versions.split(',') if v])
            for extension, versions in items
        )

@staticmethod
    def is_extensions_locked(locked_extensions, extension, version):
        # Normalise the queried name with the same rule applied at load
        # time. The previous code took a literal dict-key membership test,
        # so case- or separator-shifted variants of a locked package would
        # slip through.
        extension = ExtensionInfo._canonicalise_extension_name(extension)
        # `get(...) or []` defends against any future asymmetry between
        # membership ('in') and value lookup ('.get'): a missing/None value
        # degrades to an empty set instead of raising TypeError on set(None).
        locked_versions = locked_extensions.get(extension) or []
        return (extension in locked_extensions and (
            list(set([v.strip() for v in version.split(',') if len(version) > 0])
                & set(locked_versions)) or
            len(locked_versions) == 0
        ))

cmd: