Add support for DCE style authentication. (#63)

Adds support for DCE style authentication that is used by protocols like
RPC. DCE style authentication changes some behaviour about an
authentication protocol, e.g. Kerberos has an extra leg, and some
wrapping behaviour changes. This PR also adjusts the underlying
behaviour of `spnego.iov.BufferType.sign_only` on SSPI to represent the
buffer type of `SECBUFFER_DATA | SECBUFFER_READONLY_WITH_CHECKSUM`
rather than `SECBUFFER_MECHLIST`. This aligns the behaviour with GSSAPI
and the use of `SECBUFFER_MECHLIST` is most likely an internal flag
rather than something used publicly.

CHANGELOG.md

@@ -1,5 +1,38 @@
 # Changelog
 
+## 0.9.0 - TBD
+
+* Added the `spnego.ContextReq.dce_style` flag to enable DCE authentication mode
+  * This is used in protocols like RPC/DCE
+* The value for `spnego.iov.BufferType.sign_only` on SSPI has changed from representing `SECBUFFER_MECHLIST` to `SECBUFFER_READONLY_WITH_CHECKSUM`
+  * This is to better match what `sign_only` means when using it with GSSAPI
+  * It is needed to support RPC encryption and signature headers on SSPI
+  * The use of `SECBUFFER_MECHLIST` is not seen in any examples in the wild and is most likely an internal flag
+* Added the IOV buffer type `spnego.iov.BufferType.data_readonly`
+  * For SSPI this corresponds to `SECBUFFER_DATA | SECBUFFER_READONLY`
+  * For GSSAPI this corresponds to `GSS_IOV_BUFFER_TYPE_EMPTY`
+  * As GSSAPI has no actual equivalent to this the empty buffer type is used which in testing results in compatible buffers
+  * This is used for DCE/RPC wrapping when the PDU header and sec trailer are not signed but are included in the wrap_iov buffers.
+* Added limited support for `wrap_iov` and `unwrap_iov` in the Python NTLM context provider.
+  * This currently only supports `spnego.iov.BufferType.header`, `spnego.iov.BufferType.data`, `spnego.iov.BufferType.sign_only`, `spnego.iov.BufferType.data_readonly`, and `spnego.iov.BufferType.stream`
+  * `header`
+    * `wrap_iov`: Used to place the resulting signature in the buffer
+    * `unwrap_iov`: Used as the signature source for validation
+  * `data`
+    * `wrap_iov`: Data to be encrypted/sealed
+    * `unwrap_iov`: Data to be decrypted/unsealed
+  * `sign_only`
+    * `wrap_iov`: Data to be included in the signature/header generation
+    * `unwrap_iov`: Data to be included in the signature/header verification
+  * `data_readonly` is treated the same as `sign_only`
+  * `stream`
+    * `wrap_iov`: Not supported
+    * `unwrap_iov`: Contains the full value to decrypt with the headers in the beginning, must be coupled with a subsequent data buffer of the type `data` to place the decrypted value into
+  * The behaviour used here is modelled as closely as possible to how `SSPI` works but not all the permutations have been tested.
+  * The header/signature will be generated from the `data`, `sign_only`, `data_readonly` values concat together in the order they are provided.
+* Added the `query_message_sizes()` function on a context to retrieve the important message sizes
+  * Currently this only contains the size of the message `header`, also known as the signature or security trailer
+
 ## 0.8.0 - 2023-02-17
 
 * Added the `spnego.ContextReq.no_integrity` flag to disable integrity/confidentiality on Kerberos/Negotiate contexts

src/spnego/_context.py

@@ -2,6 +2,7 @@
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 
 import abc
+import dataclasses
 import enum
 import typing
 import warnings
@@ -112,6 +113,27 @@ class IOVUnwrapResult(typing.NamedTuple):
     qop: int  #: The Quality of Protection used for the encrypted buffers.
 
 
+@dataclasses.dataclass(frozen=True)
+class SecPkgContextSizes:
+    """Sizes of important structures used for messages.
+
+    This dataclass exposes the sizes of important structures used in message
+    support functions like wrap, wrap_iov, sign, etc. Use
+    :meth:`ContextReq.query_message_sizes` to retrieve this value for an
+    authenticated context.
+
+    Currently only ``header`` is exposed but other sizes may be added in the
+    future if needed.
+
+    Attributes:
+        header: The size of the header/signature of a wrapped token. This
+            corresponds to cbSecurityTrailer in SecPkgContext_Sizes in SSPI and
+            the size of the allocated GSS_IOV_BUFFER_TYPE_HEADER IOV buffer.
+    """
+
+    header: int
+
+
 class ContextReq(enum.IntFlag):
     none = 0x00000000
 
@@ -123,6 +145,7 @@ class ContextReq(enum.IntFlag):
     confidentiality = 0x00000010
     integrity = 0x00000020
     # anonymous = 0x00000040  # TODO: Add support for anonymous auth.
+    dce_style = 0x00001000
     identify = 0x00002000
     # Requires newer python-gssapi version to support https://github.com/pythongssapi/python-gssapi/pull/218
     delegate_policy = 0x00080000
@@ -407,6 +430,22 @@ def new_context(self) -> "ContextProxy":
         """
         pass  # pragma: no cover
 
+    @abc.abstractmethod
+    def query_message_sizes(self) -> SecPkgContextSizes:
+        """Gets the important structure sizes for message functions.
+
+        Will get the important sizes for the various message functions used by
+        the current authentication context. This must only be called once the
+        context has been authenticated.
+
+        Returns:
+            SecPkgContextSizes: The sizes for the current context.
+
+        Raises:
+            NoContextError: The security context is not ready to be queried.
+        """
+        pass  # pragma: no cover
+
     @abc.abstractmethod
     def step(
         self,

src/spnego/_credssp.py

@@ -19,6 +19,7 @@
     ContextReq,
     IOVUnwrapResult,
     IOVWrapResult,
+    SecPkgContextSizes,
     UnwrapResult,
     WinRMWrapResult,
     WrapResult,
@@ -40,6 +41,7 @@
     InvalidTokenError,
     NativeError,
     NegotiateOptions,
+    NoContextError,
     OperationNotAvailableError,
     SpnegoError,
 )
@@ -596,6 +598,15 @@ def get_extra_info(
         else:
             return default
 
+    def query_message_sizes(self) -> SecPkgContextSizes:
+        if not self._tls_object or not self.complete:
+            raise NoContextError(context_msg="Cannot get message sizes until context has been established")
+
+        cipher_negotiated, tls_protocol, _ = self._tls_object.cipher()  # type: ignore[misc]
+        trailer_length = _tls_trailer_length(0, tls_protocol, cipher_negotiated)
+
+        return SecPkgContextSizes(header=trailer_length)
+
     @_wrap_ssl_error("Invalid TLS state when wrapping data")
     def wrap(self, data: bytes, encrypt: bool = True, qop: typing.Optional[int] = None) -> WrapResult:
         self._tls_object.write(data)

src/spnego/_gss.py

@@ -14,6 +14,7 @@
     GSSMech,
     IOVUnwrapResult,
     IOVWrapResult,
+    SecPkgContextSizes,
     UnwrapResult,
     WinRMWrapResult,
     WrapResult,
@@ -59,7 +60,7 @@
 try:
     from gssapi.raw import IOV as GSSIOV
     from gssapi.raw import IOVBuffer as GSSIOVBuffer
-    from gssapi.raw import IOVBufferType, unwrap_iov, wrap_iov
+    from gssapi.raw import IOVBufferType, unwrap_iov, wrap_iov, wrap_iov_length
 except ImportError as err:
     GSSAPI_IOV_IMP_ERR = sys.exc_info()
     HAS_IOV = False
@@ -444,6 +445,19 @@ def step(
 
         return out_token
 
+    @wrap_system_error(NativeError, "Getting context sizes")
+    def query_message_sizes(self) -> SecPkgContextSizes:
+        if not self._context:
+            raise NoContextError(context_msg="Cannot get message sizes until context has been established")
+
+        iov = GSSIOV(
+            IOVBufferType.header,
+            b"",
+            std_layout=False,
+        )
+        wrap_iov_length(self._context, iov)
+        return SecPkgContextSizes(header=len(iov[0].value or b""))
+
     @wrap_system_error(NativeError, "Wrapping data")
     def wrap(self, data: bytes, encrypt: bool = True, qop: typing.Optional[int] = None) -> WrapResult:
         if not self._context:
@@ -544,6 +558,7 @@ def _context_attr_map(self) -> typing.List[typing.Tuple[ContextReq, int]]:
             (ContextReq.sequence_detect, "out_of_sequence_detection"),
             (ContextReq.confidentiality, "confidentiality"),
             (ContextReq.integrity, "integrity"),
+            (ContextReq.dce_style, "dce_style"),
             # Only present when the DCE extensions are installed.
             (ContextReq.identify, "identify"),
             # Only present with newer versions of python-gssapi https://github.com/pythongssapi/python-gssapi/pull/218.
@@ -571,4 +586,11 @@ def _convert_iov_buffer(self, buffer: IOVBuffer) -> "GSSIOVBuffer":
             auto_alloc = [BufferType.header, BufferType.padding, BufferType.trailer]
             buffer_alloc = buffer.type in auto_alloc
 
-        return GSSIOVBuffer(IOVBufferType(buffer.type), buffer_alloc, buffer_data)
+        buffer_type = buffer.type
+        if buffer.type == BufferType.data_readonly:
+            # GSSAPI doesn't have the SSPI equivalent of SECBUFFER_READONLY.
+            # the GSS_IOV_BUFFER_TYPE_EMPTY seems to produce the same behaviour
+            # so that's going to be used instead.
+            buffer_type = BufferType.empty
+
+        return GSSIOVBuffer(IOVBufferType(buffer_type), buffer_alloc, buffer_data)

src/spnego/_negotiate.py

@@ -14,6 +14,7 @@
     GSSMech,
     IOVUnwrapResult,
     IOVWrapResult,
+    SecPkgContextSizes,
     UnwrapResult,
     WinRMWrapResult,
     WrapResult,
@@ -29,7 +30,12 @@
 )
 from spnego._sspi import SSPIProxy
 from spnego.channel_bindings import GssChannelBindings
-from spnego.exceptions import BadMechanismError, InvalidTokenError, NegotiateOptions
+from spnego.exceptions import (
+    BadMechanismError,
+    InvalidTokenError,
+    NegotiateOptions,
+    NoContextError,
+)
 
 log = logging.getLogger(__name__)
 
@@ -73,7 +79,8 @@ def __init__(
         self._mech_sent = False
         self._mic_sent = False
         self._mic_recv = False
-        self._mic_required = False
+        # DCE will always send a MIC token, even for Kerberos.
+        self._mic_required = bool(self.context_req & ContextReq.dce_style)
 
     @classmethod
     def available_protocols(cls, options: typing.Optional[NegotiateOptions] = None) -> typing.List[str]:
@@ -346,6 +353,12 @@ def _step_spnego_output(
 
         return final_token
 
+    def query_message_sizes(self) -> SecPkgContextSizes:
+        if not self.complete:
+            raise NoContextError(context_msg="Cannot get message sizes until context has been established")
+
+        return self._context.query_message_sizes()
+
     def wrap(self, data: bytes, encrypt: bool = True, qop: typing.Optional[int] = None) -> WrapResult:
         return self._context.wrap(data, encrypt=encrypt, qop=qop)