codekingpro/portable-devtools

1import typing as t2 3from gssapi.raw import chan_bindings as rchan_bindings4from gssapi.raw import sec_contexts as rsec_contexts5from gssapi.raw import message as rmessage6from gssapi.raw import named_tuples as tuples7from gssapi.raw import names as rnames8from gssapi.raw import oids as roids9from gssapi.raw.types import RequirementFlag, IntEnumFlagSet10 11import gssapi.exceptions as excs12from gssapi import _utils13from gssapi.names import Name14from gssapi.creds import Credentials15 16 17class SecurityContext(rsec_contexts.SecurityContext,18                      metaclass=_utils.CheckLastError):19    """A GSSAPI Security Context20 21    This class represents a GSSAPI security context that may be used22    with and/or returned by other GSSAPI methods.23 24    It inherits from the low-level GSSAPI25    :class:`~gssapi.raw.sec_contexts.SecurityContext` class,26    and thus may used with both low-level and high-level API methods.27 28    This class may be pickled and unpickled (the attached delegated29    credentials object will not be preserved, however).30    """31 32    def __new__(33        cls,34        base: t.Optional[rsec_contexts.SecurityContext] = None,35        token: t.Optional[bytes] = None,36        name: t.Optional[rnames.Name] = None,37        creds: t.Optional[Credentials] = None,38        lifetime: t.Optional[int] = None,39        flags: t.Optional[int] = None,40        mech: t.Optional[roids.OID] = None,41        channel_bindings: t.Optional[rchan_bindings.ChannelBindings] = None,42        usage: t.Optional[str] = None,43    ) -> "SecurityContext":44 45        if token is not None:46            base = rsec_contexts.import_sec_context(token)47 48        return t.cast("SecurityContext",49                      super(SecurityContext, cls).__new__(cls, base))50 51    def __init__(52        self,53        base: t.Optional[rsec_contexts.SecurityContext] = None,54        token: t.Optional[bytes] = None,55        name: t.Optional[rnames.Name] = None,56        creds: t.Optional[Credentials] = None,57        lifetime: t.Optional[int] = None,58        flags: t.Optional[int] = None,59        mech: t.Optional[roids.OID] = None,60        channel_bindings: t.Optional[rchan_bindings.ChannelBindings] = None,61        usage: t.Optional[str] = None,62    ) -> None:63        """64        The constructor creates a new security context, but does not begin65        the initiate or accept process.66 67        If the `base` argument is used, an existing68        :class:`~gssapi.raw.sec_contexts.SecurityContext` object from69        the low-level API is converted into a high-level object.70 71        If the `token` argument is passed, the security context is imported72        using the token.73 74        Otherwise, a new security context is created.75 76        If the `usage` argument is not passed, the constructor will attempt77        to detect what the appropriate usage is based on either the existing78        security context (if `base` or `token` are used) or the argument set.79 80        For a security context of the `initiate` usage, the `name` argument81        must be used, and the `creds`, `mech`, `flags`,82        `lifetime`, and `channel_bindings` arguments may be83        used as well.84 85        For a security context of the `accept` usage, the `creds` and86        `channel_bindings` arguments may optionally be used.87        """88 89        # NB(directxman12): _last_err must be set first90        self._last_err = None91 92        # determine the usage ('initiate' vs 'accept')93        if base is None and token is None:94            # this will be a new context95            if usage is not None:96                if usage not in ('initiate', 'accept'):97                    msg = "Usage must be either 'initiate' or 'accept'"98                    raise excs.UnknownUsageError(msg, obj="security context")99 100                self.usage = usage101            elif creds is not None and creds.usage != 'both':102                self.usage = creds.usage103            elif name is not None:104                # if we pass a name, assume the usage is 'initiate'105                self.usage = 'initiate'106            else:107                # if we don't pass a name, assume the usage is 'accept'108                self.usage = 'accept'109 110            # check for appropriate arguments111            if self.usage == 'initiate':112                # takes: creds?, target_name, mech?, flags?,113                #        channel_bindings?114                if name is None:115                    raise TypeError("You must pass the 'name' argument when "116                                    "creating an initiating security context")117                self._target_name = name118                self._mech = mech119                self._desired_flags = IntEnumFlagSet(RequirementFlag, flags)120                self._desired_lifetime = lifetime121            else:122                # takes creds?123                if (name is not None or flags is not None or124                        mech is not None or lifetime is not None):125                    raise TypeError("You must pass at most the 'creds' "126                                    "argument when creating an accepting "127                                    "security context")128 129            self._channel_bindings = channel_bindings130            self._creds = creds131 132            self._delegated_creds = None133 134        else:135            # we already have a context in progress, just inspect it136            # NB(directxman12): MIT krb5 refuses to inquire about a context137            # if it's partially established, so we have to check here138 139            try:140                if self.locally_initiated:141                    self.usage = 'initiate'142                else:143                    self.usage = 'accept'144            except excs.MissingContextError:145                msg = ("Cannot extract usage from a partially completed "146                       "context")147                raise excs.UnknownUsageError(msg, obj="security context")148 149        # This is to work around an MIT krb5 bug (see the `complete` property)150        self._complete: t.Optional[bool] = None151 152    # NB(directxman12): DO NOT ADD AN __del__ TO THIS CLASS -- it screws up153    #                   the garbage collector if _last_tb is still defined154 155    # TODO(directxman12): implement flag properties156 157    def get_signature(158        self,159        message: bytes,160    ) -> bytes:161        """Calculate the signature for a message.162 163        This method calculates the signature (called a MIC) for164        the given message, which may be then used with165        :meth:`verify_signature` to confirm the validity of the166        signature.  This is useful if you wish to transmit the167        message signature and message in your own format.168 169        Args:170            message (bytes): the input message171 172        Returns:173            bytes: the message signature174 175        Raises:176            ~gssapi.exceptions.ExpiredContextError177            ~gssapi.exceptions.MissingContextError178            ~gssapi.exceptions.BadQoPError179        """180 181        # TODO(directxman12): check flags?182        return rmessage.get_mic(self, message)183 184    def verify_signature(185        self,186        message: bytes,187        mic: bytes,188    ) -> int:189        """Verify the signature for a message.190 191        This method verifies that a signature (generated by192        :meth:`get_signature` is valid for the given message.193 194        If the signature is valid, the method will return.195        Otherwise, it will raise an error.196 197        Args:198            message (bytes): the message199            mic (bytes): the signature to verify200 201        Returns:202            int: the QoP used.203 204        Raises:205            ~gssapi.exceptions.BadMICError: the signature was not valid206            ~gssapi.exceptions.InvalidTokenError207            ~gssapi.exceptions.DuplicateTokenError208            ~gssapi.exceptions.ExpiredTokenError209            ~gssapi.exceptions.TokenTooLateError210            ~gssapi.exceptions.TokenTooEarlyError211            ~gssapi.exceptions.ExpiredContextError212            ~gssapi.exceptions.MissingContextError213        """214 215        return rmessage.verify_mic(self, message, mic)216 217    def wrap(218        self,219        message: bytes,220        encrypt: bool,221    ) -> tuples.WrapResult:222        """Wrap a message, optionally with encryption223 224        This wraps a message, signing it and optionally225        encrypting it.226 227        Args:228            message (bytes): the message to wrap229            encrypt (bool): whether or not to encrypt the message230 231        Returns:232            WrapResult: the wrapped message and details about it233            (e.g. whether encryption was used succesfully)234 235        Raises:236            ~gssapi.exceptions.ExpiredContextError237            ~gssapi.exceptions.MissingContextError238            ~gssapi.exceptions.BadQoPError239        """240 241        return rmessage.wrap(self, message, encrypt)242 243    def unwrap(244        self,245        message: bytes,246    ) -> tuples.UnwrapResult:247        """Unwrap a wrapped message.248 249        This method unwraps/unencrypts a wrapped message,250        verifying the signature along the way.251 252        Args:253            message (bytes): the message to unwrap/decrypt254 255        Returns:256            UnwrapResult: the unwrapped message and details about it257            (e.g. wheter encryption was used)258 259        Raises:260            ~gssapi.exceptions.InvalidTokenError261            ~gssapi.exceptions.BadMICError262            ~gssapi.exceptions.DuplicateTokenError263            ~gssapi.exceptions.ExpiredTokenError264            ~gssapi.exceptions.TokenTooLateError265            ~gssapi.exceptions.TokenTooEarlyError266            ~gssapi.exceptions.ExpiredContextError267            ~gssapi.exceptions.MissingContextError268        """269 270        return rmessage.unwrap(self, message)271 272    def encrypt(273        self,274        message: bytes,275    ) -> bytes:276        """Encrypt a message.277 278        This method wraps and encrypts a message, similarly to279        :meth:`wrap`.  The difference is that encryption is always280        used, and the method will raise an exception if this is281        not possible.  Additionally, this method simply returns282        the encrypted message directly.283 284        Args:285            message (bytes): the message to encrypt286 287        Returns:288            bytes: the encrypted message289 290        Raises:291            ~gssapi.exceptions.EncryptionNotUsed: the encryption could not be292                used293            ~gssapi.exceptions.ExpiredContextError294            ~gssapi.exceptions.MissingContextError295            ~gssapi.exceptions.BadQoPError296        """297 298        res = self.wrap(message, encrypt=True)299 300        if not res.encrypted:301            raise excs.EncryptionNotUsed("Wrapped message was not encrypted")302 303        return res.message304 305    def decrypt(306        self,307        message: bytes,308    ) -> bytes:309        """Decrypt a message.310 311        This method decrypts and unwraps a message, verifying the signature312        along the way, similarly to :meth:`unwrap`.  The difference is that313        this method will raise an exception if encryption was established314        by the context and not used, and simply returns the decrypted315        message directly.316 317        Args:318            message (bytes): the encrypted message319 320        Returns:321            bytes: the decrypted message322 323        Raises:324            ~gssapi.exceptions.EncryptionNotUsed: encryption was expected, but325                not used326            ~gssapi.exceptions.InvalidTokenError327            ~gssapi.exceptions.BadMICError328            ~gssapi.exceptions.DuplicateTokenError329            ~gssapi.exceptions.ExpiredTokenError330            ~gssapi.exceptions.TokenTooLateError331            ~gssapi.exceptions.TokenTooEarlyError332            ~gssapi.exceptions.ExpiredContextError333            ~gssapi.exceptions.MissingContextError334        """335 336        res = self.unwrap(message)337 338        if (not res.encrypted and339                self.actual_flags & RequirementFlag.confidentiality):340            raise excs.EncryptionNotUsed("The context was established with "341                                         "encryption, but unwrapped message "342                                         "was not encrypted",343                                         unwrapped_message=res.message)344 345        return res.message346 347    def get_wrap_size_limit(348        self,349        desired_output_size: int,350        encrypted: bool = True,351    ) -> int:352        """Calculate the maximum message size for a given wrapped message size.353 354        This method calculates the maximum input message size for a given355        maximum wrapped/encrypted message size.356 357        Args:358            desired_output_size (int): the maximum output message size359            encrypted (bool): whether or not encryption should be taken360                into account361 362        Returns:363            int: the maximum input message size364 365        Raises:366            ~gssapi.exceptions.MissingContextError367            ~gssapi.exceptions.ExpiredContextError368            ~gssapi.exceptions.BadQoPError369        """370 371        return rmessage.wrap_size_limit(self, desired_output_size,372                                        encrypted)373 374    def process_token(375        self,376        token: bytes,377    ) -> None:378        """Process an output token asynchronously.379 380        This method processes an output token even when the security context381        was not expecting it.382 383        Warning:384            This method is deprecated.385 386        Args:387            token (bytes): the token to process388 389        Raises:390            ~gssapi.exceptions.InvalidTokenError391            ~gssapi.exceptions.MissingContextError392        """393 394        rsec_contexts.process_context_token(self, token)395 396    def export(self) -> bytes:397        """Export a security context.398 399        This method exports a security context, allowing it to be passed400        between processes.401 402        Returns:403            bytes: the exported security context404 405        Raises:406            ~gssapi.exceptions.ExpiredContextError407            ~gssapi.exceptions.MissingContextError408            ~gssapi.exceptions.OperationUnavailableError409        """410 411        return rsec_contexts.export_sec_context(self)412 413    _INQUIRE_ARGS = ('initiator_name', 'target_name', 'lifetime',414                     'mech', 'flags', 'locally_init', 'complete')415 416    @_utils.check_last_err417    def _inquire(418        self,419        **kwargs: bool,420    ) -> tuples.InquireContextResult:421        """Inspect the security context for information422 423        This method inspects the security context for information.424 425        If no keyword arguments are passed, all available information426        is returned.  Otherwise, only the keyword arguments that427        are passed and set to `True` are returned.428 429        Args:430            initiator_name (bool): get the initiator name for this context431            target_name (bool): get the target name for this context432            lifetime (bool): get the remaining lifetime, in seconds, for this433                context434            mech (bool): get the :class:`MechType` used by this context435            flags (bool): get the flags set on this context436            locally_init (bool): get whether this context was locally initiated437            complete (bool): get whether negotiation on this context has438                been completed439 440        Returns:441            InquireContextResult: the results of the inquiry, with unused442            fields set to None443 444        Raises:445            ~gssapi.exceptions.MissingContextError446        """447        if not kwargs:448            default_val = True449        else:450            default_val = False451 452        for arg in self._INQUIRE_ARGS:453            kwargs[arg] = kwargs.get(arg, default_val)454 455        res = rsec_contexts.inquire_context(self, **kwargs)456 457        if (kwargs.get('initiator_name', False) and458                res.initiator_name is not None):459            init_name = Name(res.initiator_name)460        else:461            init_name = None462 463        if (kwargs.get('target_name', False) and464                res.target_name is not None):465            target_name = Name(res.target_name)466        else:467            target_name = None468 469        return tuples.InquireContextResult(init_name, target_name,470                                           res.lifetime, res.mech,471                                           res.flags, res.locally_init,472                                           res.complete)473 474    @property475    def lifetime(self) -> int:476        """The amount of time for which this context remains valid"""477        return rsec_contexts.context_time(self)478 479    @property480    def delegated_creds(self) -> t.Optional[Credentials]:481        """The credentials delegated from the initiator to the acceptor482 483        .. warning::484 485            This value will not be preserved across picklings.  These should486            be separately exported and transferred.487 488        """489        return self._delegated_creds490 491    initiator_name = _utils.inquire_property(492        'initiator_name', 'The :class:`Name` of the initiator of this context')493    target_name = _utils.inquire_property(494        'target_name', 'The :class:`Name` of the target of this context')495    mech = _utils.inquire_property(496        'mech', 'The mechanism (:class:`MechType`) in use by this context')497    actual_flags = _utils.inquire_property(498        'flags', 'The flags set on this context')499    locally_initiated = _utils.inquire_property(500        'locally_init', 'Whether this context was locally intiated')501 502    @property  # type: ignore # https://github.com/python/mypy/issues/1362503    @_utils.check_last_err504    def complete(self) -> bool:505        """Whether negotiation for this context has been completed"""506        # NB(directxman12): MIT krb5 has a bug where it refuses to507        #                   inquire about partially completed contexts,508        #                   so we can't just use `self._inquire` generally509        if self._started:510            complete = self._complete511            if complete is None:512                try:513                    complete = self._inquire(complete=True).complete514                except excs.MissingContextError:515                    return False516                else:517                    self._complete = complete518 519            return complete520        else:521            return False522 523    @_utils.catch_and_return_token524    def step(525        self,526        token: t.Optional[bytes] = None,527    ) -> t.Optional[bytes]:528        """Perform a negotation step.529 530        This method performs a negotiation step based on the usage type531        of this context.  If `__DEFER_STEP_ERRORS__` is set to True on532        the class, this method will return a token, even when exceptions533        would be thrown.  The generated exception will be thrown on the next534        method call or property lookup on the context.535        **This is the default behavior.**536 537        This method should be used in a while loop, as such:538 539        .. code-block:: python540 541           input_token = None542           try:543               while not ctx.complete:544                   output_token = ctx.step(input_token)545                   if not output_token:546                       break547                   input_token = send_and_receive(output_token)548           except GSSError as e:549                handle_the_issue()550 551        .. tip::552 553            Disabling `__DEFER_STEP_ERRORS__` is rarely necessary.554            When this method is used in a loop (as above),555            `__DEFER_STEP_ERRORS__` will ensure that you always556            send an error token when it's available,557            keeping the other end of the security context updated558            with the status of the negotiation.559 560        Args:561            token (bytes): the input token from the other participant's step562 563        Returns:564            bytes: the output token to send to the other participant565 566        Raises:567            ~gssapi.exceptions.InvalidTokenError568            ~gssapi.exceptions.InvalidCredentialsError569            ~gssapi.exceptions.MissingCredentialsError570            ~gssapi.exceptions.ExpiredCredentialsError571            ~gssapi.exceptions.BadChannelBindingsError572            ~gssapi.exceptions.BadMICError573            ~gssapi.exceptions.ExpiredTokenError: (initiate only)574            ~gssapi.exceptions.DuplicateTokenError575            ~gssapi.exceptions.MissingContextError576            ~gssapi.exceptions.BadNameTypeError: (initiate only)577            ~gssapi.exceptions.BadNameError: (initiate only)578            ~gssapi.exceptions.BadMechanismError579        """580 581        if self.usage == 'accept':582            return self._acceptor_step(token=token or b"")583        else:584            return self._initiator_step(token=token)585 586    def _acceptor_step(587        self,588        token: bytes,589    ) -> t.Optional[bytes]:590        res = rsec_contexts.accept_sec_context(token, self._creds,591                                               self, self._channel_bindings)592 593        if res.delegated_creds is not None:594            self._delegated_creds = Credentials(res.delegated_creds)595        else:596            self._delegated_creds = None597 598        self._complete = not res.more_steps599 600        return res.token601 602    def _initiator_step(603        self,604        token: t.Optional[bytes] = None,605    ) -> t.Optional[bytes]:606        res = rsec_contexts.init_sec_context(self._target_name, self._creds,607                                             self, self._mech,608                                             self._desired_flags,609                                             self._desired_lifetime,610                                             self._channel_bindings,611                                             token)612 613        self._complete = not res.more_steps614 615        return res.token616 617    # pickle protocol support618    def __reduce__(619        self,620    ) -> t.Tuple[t.Type["SecurityContext"], t.Tuple[None, bytes]]:621        # the unpickle arguments to new are (base=None, token=self.export())622        return (type(self), (None, self.export()))623