Creds serialization and deserialization

src/krb5/__init__.py

@@ -172,12 +172,19 @@
 
 
 try:
-    from krb5._creds_mit import get_etype_info, get_validated_creds
+    from krb5._creds_mit import (
+        get_etype_info,
+        get_validated_creds,
+        marshal_credentials,
+        unmarshal_credentials,
+    )
 except ImportError:
     pass
 else:
     __all__.append("get_etype_info")
     __all__.append("get_validated_creds")
+    __all__.append("marshal_credentials")
+    __all__.append("unmarshal_credentials")
 
 
 try:

src/krb5/_creds.pxd

@@ -8,6 +8,7 @@ from krb5._krb5_types cimport *
 cdef class Creds:
     cdef Context ctx
     cdef krb5_creds raw
+    cdef krb5_creds* _raw_ptr
     cdef int needs_free
 
 

src/krb5/_creds.pyx

@@ -77,6 +77,11 @@ cdef extern from "python_krb5.h":
         # krb5_authdata ***authdata,
     ) nogil
 
+    void krb5_free_creds(
+        krb5_context context,
+        krb5_creds *val,
+    ) nogil
+
     void krb5_free_cred_contents(
         krb5_context context,
         krb5_creds *val,
@@ -179,15 +184,20 @@ class TicketFlags(enum.IntFlag):
 cdef class Creds:
     # cdef Context ctx
     # cdef krb5_creds raw
+    # cdef krb5_creds* _raw_ptr
     # cdef int needs_free
 
     def __cinit__(Creds self, Context context):
         self.ctx = context
         self.needs_free = 0
+        self._raw_ptr = NULL
 
     def __dealloc__(Creds self):
         if self.needs_free:
-            krb5_free_cred_contents(self.ctx.raw, &self.raw)
+            if NULL != self._raw_ptr:
+                krb5_free_creds(self.ctx.raw, self._raw_ptr)
+            else:
+                krb5_free_cred_contents(self.ctx.raw, &self.raw)
             self.needs_free = 0
 
     def __str__(Creds self) -> str:

src/krb5/_creds_mit.pyi

@@ -48,3 +48,26 @@ def get_etype_info(
 
         If there are no s2kparams in the provided etype-info, s2kparams is None.
     """
+
+def marshal_credentials(context: Context, creds: Creds) -> bytes:
+    """Serialize creds in the format used by the FILE ccache format
+    (vesion 4) and KCM ccache protocol.
+
+    Args:
+        context: Krb5 context.
+        creds: Credentials to serialize.
+
+    Returns:
+        bytes: The serialized credentials.
+    """
+
+def unmarshal_credentials(context: Context, data: bytes) -> Creds:
+    """Deserialize creds from the format used by the FILE ccache format
+    (vesion 4) and KCM ccache protocol.
+
+    Args:
+        context: Krb5 context.
+        data: serialized credentials.
+    Returns:
+        Creds: The unserialized credentials.
+    """

src/krb5/_creds_mit.pyx

@@ -6,6 +6,8 @@ import typing
 
 from krb5._exceptions import Krb5Error
 
+from libc.string cimport memcpy
+
 from krb5._ccache cimport CCache
 from krb5._context cimport Context
 from krb5._creds cimport Creds
@@ -17,7 +19,11 @@ from krb5._principal cimport Principal
 cdef extern from "python_krb5.h":
     """
 #if defined(HEIMDAL_XFREE)
-#error "Heimdal implementation of krb5_get_validated_creds() does not work"
+#error "Heimdal implementation does not support MIT-specific calls:"
+#error " krb5_get_validated_creds()"
+#error " krb5_get_etype_info()"
+#error " krb5_marshal_credentials()"
+#error " krb5_unmarshal_credentials()"
 #endif
     """
 
@@ -38,6 +44,22 @@ cdef extern from "python_krb5.h":
         krb5_data *s2kparams_ou,
     ) nogil
 
+    krb5_error_code krb5_marshal_credentials(
+        krb5_context context,
+        krb5_creds *creds,
+        krb5_data **data
+    ) nogil
+
+    krb5_error_code krb5_unmarshal_credentials(
+        krb5_context context,
+        krb5_data *data,
+        krb5_creds **creds,
+    ) nogil
+
+    void krb5_free_data(
+        krb5_context context,
+        krb5_data *val,
+    ) nogil
 
 def get_validated_creds(
     Context context not None,
@@ -120,3 +142,53 @@ EtypeInfo = collections.namedtuple('EtypeInfo', [
     'salt',
     's2kparams',
 ])
+
+def marshal_credentials(Context context not None, Creds creds not None) -> bytes:
+    cdef krb5_error_code err = 0
+    cdef krb5_data *data = NULL
+    cdef size_t length
+    cdef char *value
+
+    try:
+        err = krb5_marshal_credentials(context.raw, &creds.raw, &data)
+
+        if err:
+            raise Krb5Error(context, err)
+
+        pykrb5_get_krb5_data(data, &length, &value)
+
+        if length == 0:
+            data_bytes = b""
+        else:
+            data_bytes = value[:length]
+
+        return data_bytes
+
+    finally:
+        if NULL != data:
+            krb5_free_data(context.raw, data)
+
+def unmarshal_credentials(Context context not None, const unsigned char[:] data not None) -> Creds:
+    cdef krb5_error_code err = 0
+    cdef krb5_data data_raw
+
+    creds = Creds(context)
+
+    if len(data) == 0:
+        pykrb5_set_krb5_data(&data_raw, 0, "")
+    else:
+        pykrb5_set_krb5_data(&data_raw, len(data), <char *>&data[0])
+
+    err = krb5_unmarshal_credentials(context.raw, &data_raw, &creds._raw_ptr)
+
+    if creds._raw_ptr:
+        # creds_raw was calloc'ed for krb5_creds structure
+        # "shallow copy" the structure
+        memcpy(&creds.raw, creds._raw_ptr, sizeof(creds.raw))
+        # mark "deep copy" for future deallocation
+        creds.needs_free = 1
+
+    if err:
+        raise Krb5Error(context, err)
+
+    return creds

tests/test_creds.py

@@ -292,3 +292,31 @@ def test_get_etype_info(realm: k5test.K5Realm, tmp_path: pathlib.Path) -> None:
     # Verify the credentials
     assert creds.client.name == realm.user_princ.encode()
     assert creds.server.name == b"krbtgt/KRBTEST.COM@KRBTEST.COM"
+
+
+@pytest.mark.requires_api("marshal_credentials")
+def test_creds_serialization(realm: k5test.K5Realm) -> None:
+    ctx = krb5.init_context()
+    princ = krb5.parse_name_flags(ctx, realm.user_princ.encode())
+    opt = krb5.get_init_creds_opt_alloc(ctx)
+    creds = krb5.get_init_creds_password(ctx, princ, opt, realm.password("user").encode())
+    assert isinstance(creds, krb5.Creds)
+
+    with pytest.raises(krb5.Krb5Error):
+        krb5.unmarshal_credentials(ctx, b"invalid")
+
+    with pytest.raises(krb5.Krb5Error):
+        krb5.unmarshal_credentials(ctx, b"")
+
+    bytes = krb5.marshal_credentials(ctx, creds)
+    assert len(bytes) > 0
+
+    uncreds = krb5.unmarshal_credentials(ctx, bytes)
+    assert isinstance(uncreds, krb5.Creds)
+    assert str(uncreds) == "Creds"
+
+    assert id(creds) != id(uncreds)
+    assert creds.client.name == uncreds.client.name
+    assert creds.ticket == uncreds.ticket
+    assert creds.keyblock.data == uncreds.keyblock.data
+    assert creds.times.endtime == uncreds.times.endtime