Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes14kdownloads
creds.py477 linesDownload Raw Back to gssapi
1import typing as t2 3from gssapi.raw import creds as rcreds4from gssapi.raw import named_tuples as tuples5from gssapi.raw import names as rnames6from gssapi.raw import oids as roids7from gssapi._utils import import_gssapi_extension, _encode_dict8 9from gssapi import names10 11rcred_imp_exp = import_gssapi_extension('cred_imp_exp')12rcred_s4u = import_gssapi_extension('s4u')13rcred_cred_store = import_gssapi_extension('cred_store')14rcred_rfc5588 = import_gssapi_extension('rfc5588')15 16 17class Credentials(rcreds.Creds):18    """GSSAPI Credentials19 20    This class represents a set of GSSAPI credentials which may21    be used with and/or returned by other GSSAPI methods.22 23    It inherits from the low-level GSSAPI :class:`~gssapi.raw.creds.Creds`24    class, and thus may used with both low-level and high-level API methods.25 26    If your implementation of GSSAPI supports the credentials import-export27    extension, you may pickle and unpickle this object.28 29    The constructor either acquires or imports a set of GSSAPI30    credentials.31 32    If the `base` argument is used, an existing33    :class:`~gssapi.raw.creds.Creds` object from the low-level API is34    converted into a high-level object.35 36    If the `token` argument is used, the credentials37    are imported using the token, if the credentials import-export38    extension is supported (:requires-ext:`cred_imp_exp`).39 40    Otherwise, the credentials are acquired as per the41    :meth:`acquire` method.42 43    Raises:44        ~gssapi.exceptions.BadMechanismError45        ~gssapi.exceptions.BadNameTypeError46        ~gssapi.exceptions.BadNameError47        ~gssapi.exceptions.ExpiredCredentialsError48        ~gssapi.exceptions.MissingCredentialsError49    """50 51    __slots__ = ()52 53    def __new__(54        cls,55        base: t.Optional[rcreds.Creds] = None,56        token: t.Optional[bytes] = None,57        name: t.Optional[rnames.Name] = None,58        lifetime: t.Optional[int] = None,59        mechs: t.Optional[t.Iterable[roids.OID]] = None,60        usage: str = 'both',61        store: t.Optional[62            t.Dict[t.Union[bytes, str], t.Union[bytes, str]]63        ] = None,64    ) -> "Credentials":65        # TODO(directxman12): this is missing support for password66        #                     (non-RFC method)67        if base is not None:68            base_creds = base69        elif token is not None:70            if rcred_imp_exp is None:71                raise NotImplementedError("Your GSSAPI implementation does "72                                          "not have support for importing and "73                                          "exporting creditials")74 75            base_creds = rcred_imp_exp.import_cred(token)76        else:77            res = cls.acquire(name, lifetime, mechs, usage,78                              store=store)79            base_creds = res.creds80 81        return t.cast("Credentials",82                      super(Credentials, cls).__new__(cls, base_creds))83 84    @property85    def name(self) -> rnames.Name:86        """Get the name associated with these credentials"""87        return t.cast(rnames.Name,88                      self.inquire(name=True, lifetime=False, usage=False,89                                   mechs=False).name)90 91    @property92    def lifetime(self) -> int:93        """Get the remaining lifetime of these credentials, in seconds"""94        return t.cast(int,95                      self.inquire(name=False, lifetime=True,96                                   usage=False, mechs=False).lifetime)97 98    @property99    def mechs(self) -> t.Set[roids.OID]:100        """Get the mechanisms for these credentials"""101        return t.cast(t.Set[roids.OID],102                      self.inquire(name=False, lifetime=False,103                                   usage=False, mechs=True).mechs)104 105    @property106    def usage(self) -> str:107        """Get the usage (initiate, accept, or both) of these credentials"""108        return t.cast(str,109                      self.inquire(name=False, lifetime=False,110                                   usage=True, mechs=False).usage)111 112    @classmethod113    def acquire(114        cls,115        name: t.Optional[rnames.Name] = None,116        lifetime: t.Optional[int] = None,117        mechs: t.Optional[t.Iterable[roids.OID]] = None,118        usage: str = 'both',119        store: t.Optional[120            t.Dict[t.Union[bytes, str], t.Union[bytes, str]]121        ] = None,122    ) -> tuples.AcquireCredResult:123        """Acquire GSSAPI credentials124 125        This method acquires credentials.  If the `store` argument is126        used, the credentials will be acquired from the given127        credential store (if supported).  Otherwise, the credentials are128        acquired from the default store.129 130        The credential store information is a dictionary containing131        mechanisms-specific keys and values pointing to a credential store132        or stores.133 134        Using a non-default store requires support for the credentials store135        extension.136 137        Args:138            name (~gssapi.names.Name): the name associated with the139                credentials, or None for the default name140            lifetime (int): the desired lifetime of the credentials in seconds,141                or None for indefinite142            mechs (list): the desired :class:`MechType` OIDs to be used143                with the credentials, or None for the default set144            usage (str): the usage for the credentials -- either 'both',145                'initiate', or 'accept'146            store (dict): the credential store information pointing to the147                credential store from which to acquire the credentials,148                or None for the default store (:requires-ext:`cred_store`)149 150        Returns:151            AcquireCredResult: the acquired credentials and information about152            them153 154        Raises:155            ~gssapi.exceptions.BadMechanismError156            ~gssapi.exceptions.BadNameTypeError157            ~gssapi.exceptions.BadNameError158            ~gssapi.exceptions.ExpiredCredentialsError159            ~gssapi.exceptions.MissingCredentialsError160        """161 162        if store is None:163            res = rcreds.acquire_cred(name, lifetime,164                                      mechs, usage)165        else:166            if rcred_cred_store is None:167                raise NotImplementedError("Your GSSAPI implementation does "168                                          "not have support for manipulating "169                                          "credential stores")170 171            b_store = _encode_dict(store)172 173            res = rcred_cred_store.acquire_cred_from(b_store, name,174                                                     lifetime, mechs,175                                                     usage)176 177        return tuples.AcquireCredResult(cls(base=res.creds), res.mechs,178                                        res.lifetime)179 180    def store(181        self,182        store: t.Optional[183            t.Dict[t.Union[bytes, str], t.Union[bytes, str]]184        ] = None,185        usage: str = 'both',186        mech: t.Optional[roids.OID] = None,187        overwrite: bool = False,188        set_default: bool = False,189    ) -> tuples.StoreCredResult:190        """Store these credentials into the given store191 192        This method stores the current credentials into the specified193        credentials store.  If the default store is used, support for194        :rfc:`5588` is required.  Otherwise, support for the credentials195        store extension is required.196 197        :requires-ext:`rfc5588` or :requires-ext:`cred_store`198 199        Args:200            store (dict): the store into which to store the credentials,201                or None for the default store.202            usage (str): the usage to store the credentials with -- either203                'both', 'initiate', or 'accept'204            mech (~gssapi.OID): the :class:`MechType` to associate with the205                stored credentials206            overwrite (bool): whether or not to overwrite existing credentials207                stored with the same name, etc208            set_default (bool): whether or not to set these credentials as209                the default credentials for the given store.210 211        Returns:212            StoreCredResult: the results of the credential storing operation213 214        Raises:215            ~gssapi.exceptions.GSSError216            ~gssapi.exceptions.ExpiredCredentialsError217            ~gssapi.exceptions.MissingCredentialsError218            ~gssapi.exceptions.OperationUnavailableError219            ~gssapi.exceptions.DuplicateCredentialsElementError220        """221 222        if store is None:223            if rcred_rfc5588 is None:224                raise NotImplementedError("Your GSSAPI implementation does "225                                          "not have support for RFC 5588")226 227            return rcred_rfc5588.store_cred(self, usage, mech,228                                            overwrite, set_default)229        else:230            if rcred_cred_store is None:231                raise NotImplementedError("Your GSSAPI implementation does "232                                          "not have support for manipulating "233                                          "credential stores directly")234 235            b_store = _encode_dict(store)236 237            return rcred_cred_store.store_cred_into(b_store, self, usage, mech,238                                                    overwrite, set_default)239 240    def impersonate(241        self,242        name: t.Optional[rnames.Name] = None,243        lifetime: t.Optional[int] = None,244        mechs: t.Optional[t.Iterable[roids.OID]] = None,245        usage: str = 'initiate',246    ) -> "Credentials":247        """Impersonate a name using the current credentials248 249        This method acquires credentials by impersonating another250        name using the current credentials.251 252        :requires-ext:`s4u`253 254        Args:255            name (~gssapi.names.Name): the name to impersonate256            lifetime (int): the desired lifetime of the new credentials in257                seconds, or None for indefinite258            mechs (list): the desired :class:`MechType` OIDs for the new259                credentials260            usage (str): the desired usage for the new credentials -- either261                'both', 'initiate', or 'accept'.  Note that some mechanisms262                may only support 'initiate'.263 264        Returns:265            Credentials: the new credentials impersonating the given name266        """267 268        if rcred_s4u is None:269            raise NotImplementedError("Your GSSAPI implementation does not "270                                      "have support for S4U")271 272        res = rcred_s4u.acquire_cred_impersonate_name(self, name,273                                                      lifetime, mechs,274                                                      usage)275 276        return type(self)(base=res.creds)277 278    def inquire(279        self,280        name: bool = True,281        lifetime: bool = True,282        usage: bool = True,283        mechs: bool = True,284    ) -> tuples.InquireCredResult:285        """Inspect these credentials for information286 287        This method inspects these credentials for information about them.288 289        Args:290            name (bool): get the name associated with the credentials291            lifetime (bool): get the remaining lifetime for the credentials292            usage (bool): get the usage for the credentials293            mechs (bool): get the mechanisms associated with the credentials294 295        Returns:296            InquireCredResult: the information about the credentials,297            with None used when the corresponding argument was False298 299        Raises:300            ~gssapi.exceptions.MissingCredentialsError301            ~gssapi.exceptions.InvalidCredentialsError302            ~gssapi.exceptions.ExpiredCredentialsError303        """304 305        res = rcreds.inquire_cred(self, name, lifetime, usage, mechs)306 307        if res.name is not None:308            res_name = names.Name(res.name)309        else:310            res_name = None311 312        return tuples.InquireCredResult(res_name, res.lifetime,313                                        res.usage, res.mechs)314 315    def inquire_by_mech(316        self,317        mech: roids.OID,318        name: bool = True,319        init_lifetime: bool = True,320        accept_lifetime: bool = True,321        usage: bool = True,322    ) -> tuples.InquireCredByMechResult:323        """Inspect these credentials for per-mechanism information324 325        This method inspects these credentials for per-mechanism information326        about them.327 328        Args:329            mech (~gssapi.OID): the mechanism for which to retrieve the330                information331            name (bool): get the name associated with the credentials332            init_lifetime (bool): get the remaining initiate lifetime for333                the credentials in seconds334            accept_lifetime (bool): get the remaining accept lifetime for335                the credentials in seconds336            usage (bool): get the usage for the credentials337 338        Returns:339            InquireCredByMechResult: the information about the credentials,340            with None used when the corresponding argument was False341        """342 343        res = rcreds.inquire_cred_by_mech(self, mech, name, init_lifetime,344                                          accept_lifetime, usage)345 346        if res.name is not None:347            res_name = names.Name(res.name)348        else:349            res_name = None350 351        return tuples.InquireCredByMechResult(res_name,352                                              res.init_lifetime,353                                              res.accept_lifetime,354                                              res.usage)355 356    def add(357        self,358        name: rnames.Name,359        mech: roids.OID,360        usage: str = 'both',361        init_lifetime: t.Optional[int] = None,362        accept_lifetime: t.Optional[int] = None,363        impersonator: t.Optional[rcreds.Creds] = None,364        store: t.Optional[365            t.Dict[t.Union[bytes, str], t.Union[bytes, str]]366        ] = None,367    ) -> "Credentials":368        """Acquire more credentials to add to the current set369 370        This method works like :meth:`acquire`, except that it adds the371        acquired credentials for a single mechanism to a copy of the current372        set, instead of creating a new set for multiple mechanisms.373        Unlike :meth:`acquire`, you cannot pass None desired name or374        mechanism.375 376        If the `impersonator` argument is used, the credentials will377        impersonate the given name using the impersonator credentials378        (:requires-ext:`s4u`).379 380        If the `store` argument is used, the credentials will be acquired381        from the given credential store (:requires-ext:`cred_store`).382        Otherwise, the credentials are acquired from the default store.383 384        The credential store information is a dictionary containing385        mechanisms-specific keys and values pointing to a credential store386        or stores.387 388        Note that the `store` argument is not compatible with the389        `impersonator` argument.390 391        Args:392            name (~gssapi.names.Name): the name associated with the393                credentials394            mech (~gssapi.OID): the desired :class:`MechType` to be used with395                the credentials396            usage (str): the usage for the credentials -- either 'both',397                'initiate', or 'accept'398            init_lifetime (int): the desired initiate lifetime of the399                credentials in seconds, or None for indefinite400            accept_lifetime (int): the desired accept lifetime of the401                credentials in seconds, or None for indefinite402            impersonator (Credentials): the credentials to use to impersonate403                the given name, or None to not acquire normally404                (:requires-ext:`s4u`)405            store (dict): the credential store information pointing to the406                credential store from which to acquire the credentials,407                or None for the default store (:requires-ext:`cred_store`)408 409        Returns:410            Credentials: the credentials set containing the current credentials411            and the newly acquired ones.412 413        Raises:414            ~gssapi.exceptions.BadMechanismError415            ~gssapi.exceptions.BadNameTypeError416            ~gssapi.exceptions.BadNameError417            ~gssapi.exceptions.DuplicateCredentialsElementError418            ~gssapi.exceptions.ExpiredCredentialsError419            ~gssapi.exceptions.MissingCredentialsError420        """421 422        if store is not None and impersonator is not None:423            raise ValueError('You cannot use both the `impersonator` and '424                             '`store` arguments at the same time')425 426        if store is not None:427            if rcred_cred_store is None:428                raise NotImplementedError("Your GSSAPI implementation does "429                                          "not have support for manipulating "430                                          "credential stores")431            b_store = _encode_dict(store)432 433            res = rcred_cred_store.add_cred_from(b_store, self, name, mech,434                                                 usage, init_lifetime,435                                                 accept_lifetime)436        elif impersonator is not None:437            if rcred_s4u is None:438                raise NotImplementedError("Your GSSAPI implementation does "439                                          "not have support for S4U")440            res = rcred_s4u.add_cred_impersonate_name(self, impersonator,441                                                      name, mech, usage,442                                                      init_lifetime,443                                                      accept_lifetime)444        else:445            res = rcreds.add_cred(self, name, mech, usage, init_lifetime,446                                  accept_lifetime)447 448        return Credentials(res.creds)449 450    def export(self) -> bytes:451        """Export these credentials into a token452 453        This method exports the current credentials to a token that can454        then be imported by passing the `token` argument to the constructor.455 456        This is often used to pass credentials between processes.457 458        :requires-ext:`cred_imp_exp`459 460        Returns:461            bytes: the exported credentials in token form462        """463 464        if rcred_imp_exp is None:465            raise NotImplementedError("Your GSSAPI implementation does not "466                                      "have support for importing and "467                                      "exporting creditials")468 469        return rcred_imp_exp.export_cred(self)470 471    # pickle protocol support472    def __reduce__(473        self,474    ) -> t.Tuple[t.Type["Credentials"], t.Tuple[None, bytes]]:475        # the unpickle arguments to new are (base=None, token=self.export())476        return (type(self), (None, self.export()))477 
codekingpro/portable-devtools · Team Ai