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,
+        serialize_creds,
+        unserialize_creds,
+    )
 except ImportError:
     pass
 else:
     __all__.append("get_etype_info")
     __all__.append("get_validated_creds")
+    __all__.append("serialize_creds")
+    __all__.append("unserialize_creds")
 
 
 try:

src/krb5/_creds_mit.pyi

@@ -48,3 +48,25 @@ def get_etype_info(
 
         If there are no s2kparams in the provided etype-info, s2kparams is None.
     """
+
+def serialize_creds(context: Context, creds: Creds) -> bytes:
+    """Serialize creds in the format used by the FILE ccache format
+    (vesion 4) and KCM ccache protocol.
+
+    Args:
+        creds: Credentials to serialize.
+
+    Returns:
+        bytes: The serialized credentials.
+    """
+
+def unserialize_creds(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,9 @@ import typing
 
 from krb5._exceptions import Krb5Error
 
+from libc.stdlib cimport free
+from libc.string cimport memcpy
+
 from krb5._ccache cimport CCache
 from krb5._context cimport Context
 from krb5._creds cimport Creds
@@ -17,7 +20,19 @@ 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()"
+#else
+    void pykrb5_free_data(
+        krb5_context context,
+        krb5_data *val
+    )
+    {
+        krb5_free_data(context, val);
+    }
 #endif
     """
 
@@ -38,6 +53,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 pykrb5_free_data(
+        krb5_context context,
+        krb5_data *val,
+    ) nogil
 
 def get_validated_creds(
     Context context not None,
@@ -120,3 +151,55 @@ EtypeInfo = collections.namedtuple('EtypeInfo', [
     'salt',
     's2kparams',
 ])
+
+def serialize_creds(Context context not None, Creds self 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, &self.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:
+            pykrb5_free_data(context.raw, data)
+
+def unserialize_creds(Context context not None, const unsigned char[:] data not None) -> Creds:
+    cdef krb5_error_code err = 0
+    cdef krb5_creds *creds_raw = NULL
+    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)
+
+    if creds_raw:
+        # creds_raw was calloc'ed for krb5_creds structure
+        # "shallow copy" the structure and free it
+        memcpy(&creds.raw, creds_raw, sizeof(creds.raw))
+        free(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,29 @@ 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("serialize_creds")
+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.unserialize_creds(ctx, b"invalid")
+
+    with pytest.raises(krb5.Krb5Error):
+        krb5.unserialize_creds(ctx, b"")
+
+    bytes = krb5.serialize_creds(ctx, creds)
+    assert len(bytes) > 0
+
+    uncreds = krb5.unserialize_creds(ctx, bytes)
+    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