1#@!#!123s

: / opt / cloudlinux / venv / lib64 / python3.11 / site-packages / clselect / clselectpythonuser /
Filename : extensions.py
back
# -*- 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 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 = ()

        return dict((extension, [v.strip() for v in versions.split(',') if v])
                    for extension, versions in items)

    @staticmethod
    def is_extensions_locked(locked_extensions, extension, version):
        return (extension in locked_extensions and (
            list(set([v.strip() for v in version.split(',') if len(version) > 0])
                & set(locked_extensions.get(extension))) or
            len(locked_extensions.get(extension)) == 0
        ))