codekingpro/portable-devtools

1import sys2import types3import typing as t4 5import decorator as deco6 7from gssapi.raw.misc import GSSError8 9if t.TYPE_CHECKING:10    from gssapi.sec_contexts import SecurityContext11 12 13def import_gssapi_extension(14    name: str,15) -> t.Optional[types.ModuleType]:16    """Import a GSSAPI extension module17 18    This method imports a GSSAPI extension module based19    on the name of the extension (not including the20    'ext_' prefix).  If the extension is not available,21    the method retuns None.22 23    Args:24        name (str): the name of the extension25 26    Returns:27        module: Either the extension module or None28    """29 30    try:31        path = 'gssapi.raw.ext_{0}'.format(name)32        __import__(path)33        return sys.modules[path]34    except ImportError:35        return None36 37 38def inquire_property(39    name: str,40    doc: t.Optional[str] = None41) -> property:42    """Creates a property based on an inquire result43 44    This method creates a property that calls the45    :python:`_inquire` method, and return the value of the46    requested information.47 48    Args:49        name (str): the name of the 'inquire' result information50 51    Returns:52        property: the created property53    """54 55    def inquire_property(self: "SecurityContext") -> t.Any:56        if not self._started:57            msg = (f"Cannot read {name} from a security context whose "58                   "establishment has not yet been started.")59            raise AttributeError(msg)60 61        return getattr(self._inquire(**{name: True}), name)62 63    return property(inquire_property, doc=doc)64 65 66# use UTF-8 as the default encoding, like Python 367_ENCODING = 'UTF-8'68 69 70def _get_encoding() -> str:71    """Gets the current encoding used for strings.72 73    This value is used to encode and decode string74    values like names.75 76    Returns:77        str: the current encoding78    """79    return _ENCODING80 81 82def set_encoding(83    enc: str,84) -> None:85    """Sets the current encoding used for strings86 87    This value is used to encode and decode string88    values like names.89 90    Args:91        enc: the encoding to use92    """93 94    global _ENCODING95    _ENCODING = enc96 97 98def _encode_dict(99    d: t.Dict[t.Union[bytes, str], t.Union[bytes, str]],100) -> t.Dict[bytes, bytes]:101    """Encodes any relevant strings in a dict"""102    def enc(x: t.Union[bytes, str]) -> bytes:103        if isinstance(x, str):104            return x.encode(_ENCODING)105        else:106            return x107 108    return {enc(k): enc(v) for k, v in d.items()}109 110 111# in case of Python 3, just use exception chaining112@deco.decorator113def catch_and_return_token(114    func: t.Callable,115    self: "SecurityContext",116    *args: t.Any,117    **kwargs: t.Any,118) -> t.Optional[bytes]:119    """Optionally defer exceptions and return a token instead120 121    When `__DEFER_STEP_ERRORS__` is set on the implementing class122    or instance, methods wrapped with this wrapper will123    catch and save their :python:`GSSError` exceptions and124    instead return the result token attached to the exception.125 126    The exception can be later retrived through :python:`_last_err`127    (and :python:`_last_tb` when Python 2 is in use).128    """129 130    try:131        return func(self, *args, **kwargs)132    except GSSError as e:133        defer_step_errors = getattr(self, '__DEFER_STEP_ERRORS__', False)134        if e.token is not None and defer_step_errors:135            self._last_err = e136            # skip the "return func" line above in the traceback137            tb = e.__traceback__.tb_next  # type: ignore[union-attr]138            self._last_err.__traceback__ = tb139 140            return e.token141        else:142            raise143 144 145@deco.decorator146def check_last_err(147    func: t.Callable,148    self: "SecurityContext",149    *args: t.Any,150    **kwargs: t.Any,151) -> t.Any:152    """Check and raise deferred errors before running the function153 154    This method checks :python:`_last_err` before running the wrapped155    function.  If present and not None, the exception will be raised156    with its original traceback.157    """158 159    if self._last_err is not None:160        try:161            raise self._last_err162        finally:163            self._last_err = None164    else:165        return func(self, *args, **kwargs)166 167 168class CheckLastError(type):169    """Check for a deferred error on all methods170 171    This metaclass applies the :python:`check_last_err` decorator172    to all methods not prefixed by '_'.173 174    Additionally, it enabled `__DEFER_STEP_ERRORS__` by default.175    """176 177    def __new__(178        cls,179        name: str,180        parents: t.Tuple[t.Type],181        attrs: t.Dict[str, t.Any],182    ) -> "CheckLastError":183        attrs['__DEFER_STEP_ERRORS__'] = True184 185        for attr_name in attrs:186            attr = attrs[attr_name]187 188            # wrap only methods189            if not isinstance(attr, types.FunctionType):190                continue191 192            if attr_name[0] != '_':193                attrs[attr_name] = check_last_err(attr)194 195        return super(CheckLastError, cls).__new__(cls, name, parents, attrs)196