Add krb5_get_renewed_creds() and krb5_get_validated_creds()

setup.py

@@ -220,6 +220,7 @@ def get_krb5_lib_path(
         "context",
         ("context_mit", "krb5_init_secure_context"),
         "creds",
+        ("creds_mit", "krb5_get_etype_info"),
         "creds_opt",
         ("creds_opt_heimdal", "krb5_get_init_creds_opt_set_default_flags"),
         ("creds_opt_mit", "krb5_get_init_creds_opt_set_out_ccache"),

src/krb5/__init__.py

@@ -38,6 +38,7 @@
     TicketTimes,
     get_init_creds_keytab,
     get_init_creds_password,
+    get_renewed_creds,
     init_creds_get,
     init_creds_get_creds,
     init_creds_init,
@@ -129,6 +130,7 @@
     "get_init_creds_opt_set_salt",
     "get_init_creds_opt_set_tkt_life",
     "get_init_creds_password",
+    "get_renewed_creds",
     "init_context",
     "init_creds_get",
     "init_creds_get_creds",
@@ -163,6 +165,14 @@
     __all__.append("cc_dup")
 
 
+try:
+    from krb5._creds_mit import get_validated_creds
+except ImportError:
+    pass
+else:
+    __all__.append("get_validated_creds")
+
+
 try:
     from krb5._ccache_match import cc_cache_match
 except ImportError:

src/krb5/_creds.pyi

@@ -3,6 +3,7 @@
 
 import typing
 
+from krb5._ccache import CCache
 from krb5._context import Context
 from krb5._creds_opt import GetInitCredsOpt
 from krb5._keyblock import KeyBlock
@@ -247,3 +248,18 @@ def init_creds_set_password(
         ctx: Initial credentials context.
         password: The password to set.
     """
+
+def get_renewed_creds(
+    context: Context,
+    client: Principal,
+    ccache: CCache,
+    in_tkt_service: typing.Optional[bytes] = None,
+) -> Creds:
+    """Get renewed credentials from the KDC.
+
+    Args:
+        context: Krb5 context.
+        client: Client principal name.
+        ccache: The cache to get the existing credentials from.
+        in_tkt_service: Server principal string or None.
+    """

src/krb5/_creds.pyx

@@ -8,6 +8,7 @@ from krb5._exceptions import Krb5Error
 from krb5._keyblock import copy_keyblock
 from krb5._principal import copy_principal
 
+from krb5._ccache cimport CCache
 from krb5._context cimport Context
 from krb5._creds_opt cimport GetInitCredsOpt
 from krb5._keyblock cimport KeyBlock
@@ -129,6 +130,14 @@ cdef extern from "python_krb5.h":
         const char *password,
     ) nogil
 
+    krb5_error_code krb5_get_renewed_creds(
+        krb5_context context,
+        krb5_creds *creds,
+        krb5_principal client,
+        krb5_ccache ccache,
+        const char *in_tkt_service,
+    ) nogil
+
 
 cdef class Creds:
     # cdef Context ctx
@@ -466,6 +475,26 @@ def init_creds_set_password(
     if err:
         raise Krb5Error(context, err)
 
+def get_renewed_creds(
+    Context context not None,
+    Principal client not None,
+    CCache ccache not None,
+    const unsigned char[:] in_tkt_service = None,
+) -> Creds:
+    creds = Creds(context)
+    cdef krb5_error_code err = 0
+
+    cdef const char *in_tkt_service_ptr = NULL
+    if in_tkt_service is not None and len(in_tkt_service):
+        in_tkt_service_ptr = <const char*>&in_tkt_service[0]
+
+    err = krb5_get_renewed_creds(context.raw, &creds.raw, client.raw, ccache.raw, in_tkt_service_ptr)
+    if err:
+        raise Krb5Error(context, err)
+
+    creds.needs_free = 1
+
+    return creds
 
 TicketTimes = collections.namedtuple('TicketTimes', [
     'authtime',

src/krb5/_creds_mit.pyi

@@ -0,0 +1,24 @@
+# Copyright: (c) 2021 Jordan Borean (@jborean93) <jborean93@gmail.com>
+# MIT License (see LICENSE or https://opensource.org/licenses/MIT)
+
+import typing
+
+from krb5._ccache import CCache
+from krb5._context import Context
+from krb5._creds import Creds
+from krb5._principal import Principal
+
+def get_validated_creds(
+    context: Context,
+    client: Principal,
+    ccache: CCache,
+    in_tkt_service: typing.Optional[bytes] = None,
+) -> Creds:
+    """Get validated credentials from the KDC for a postdated ticket.
+
+    Args:
+        context: Krb5 context.
+        client: Client principal name.
+        ccache: The cache to get the existing credentials from.
+        in_tkt_service: Server principal string or None.
+    """

src/krb5/_creds_mit.pyx

@@ -0,0 +1,50 @@
+# Copyright: (c) 2021 Jordan Borean (@jborean93) <jborean93@gmail.com>
+# MIT License (see LICENSE or https://opensource.org/licenses/MIT)
+
+import typing
+
+from krb5._exceptions import Krb5Error
+
+from krb5._ccache cimport CCache
+from krb5._context cimport Context
+from krb5._creds cimport Creds
+from krb5._krb5_types cimport *
+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"
+#endif
+    """
+
+    krb5_error_code krb5_get_validated_creds(
+        krb5_context context,
+        krb5_creds *creds,
+        krb5_principal client,
+        krb5_ccache ccache,
+        const char *in_tkt_service,
+    ) nogil
+
+
+def get_validated_creds(
+    Context context not None,
+    Principal client not None,
+    CCache ccache not None,
+    const unsigned char[:] in_tkt_service = None,
+) -> Creds:
+    creds = Creds(context)
+    cdef krb5_error_code err = 0
+
+    cdef const char *in_tkt_service_ptr = NULL
+    if in_tkt_service is not None and len(in_tkt_service):
+        in_tkt_service_ptr = <const char*>&in_tkt_service[0]
+
+    err = krb5_get_validated_creds(context.raw, &creds.raw, client.raw, ccache.raw, in_tkt_service_ptr)
+    if err:
+        raise Krb5Error(context, err)
+
+    creds.needs_free = 1
+
+    return creds

tests/test_creds.py

@@ -1,6 +1,7 @@
 # Copyright: (c) 2021 Jordan Borean (@jborean93) <jborean93@gmail.com>
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 
+import time
 import typing
 
 import k5test
@@ -154,3 +155,67 @@ def test_init_creds_set_password_invalid(realm: k5test.K5Realm) -> None:
     # Too many different error messages - just expect an error
     with pytest.raises(krb5.Krb5Error):
         krb5.init_creds_get(ctx, creds_ctx)
+
+
+def test_renew_creds(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)
+    # Ask for a renewable ticket
+    krb5.get_init_creds_opt_set_renew_life(opt, 1024)
+    creds = krb5.get_init_creds_password(ctx, princ, opt, realm.password("user").encode())
+
+    assert creds.client.name == realm.user_princ.encode()
+    assert creds.server.name == b"krbtgt/KRBTEST.COM@KRBTEST.COM"
+
+    cc = krb5.cc_new_unique(ctx, b"MEMORY")
+    krb5.cc_initialize(ctx, cc, princ)
+    krb5.cc_store_cred(ctx, cc, creds)
+
+    new_creds = krb5.get_renewed_creds(ctx, creds.client, cc)
+    assert new_creds.client.name == realm.user_princ.encode()
+    assert new_creds.server.name == b"krbtgt/KRBTEST.COM@KRBTEST.COM"
+
+    new_creds = krb5.get_renewed_creds(ctx, creds.client, cc, b"krbtgt/KRBTEST.COM@KRBTEST.COM")
+    assert new_creds.client.name == realm.user_princ.encode()
+    assert new_creds.server.name == b"krbtgt/KRBTEST.COM@KRBTEST.COM"
+
+
+@pytest.mark.requires_api("get_validated_creds")
+def test_validate_creds(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)
+    # Get postdated ticket, ticket will be valid after 1s
+    creds = krb5.get_init_creds_password(ctx, princ, opt, realm.password("user").encode(), start_time=1)
+    # Ticket flags for creds should have TKT_FLG_POSTDATED and TKT_FLG_INVALID set
+
+    assert creds.client.name == realm.user_princ.encode()
+    assert creds.server.name == b"krbtgt/KRBTEST.COM@KRBTEST.COM"
+
+    cc = krb5.cc_new_unique(ctx, b"MEMORY")
+    krb5.cc_initialize(ctx, cc, princ)
+    krb5.cc_store_cred(ctx, cc, creds)
+
+    start_time = time.time()
+    while True:
+        try:
+            new_creds = krb5.get_validated_creds(ctx, creds.client, cc)
+            break
+        except krb5.Krb5Error as e:
+            # Retry within the first 5s when the error is
+            # KRB5KRB_AP_ERR_TKT_NYV ("Ticket not yet valid"). The ticket should
+            # normally be valid after 0-1s.
+            if (time.time() - start_time < 5) and e.err_code == -1765328351:
+                # Retry
+                time.sleep(0.1)
+            else:
+                raise
+    assert new_creds.client.name == realm.user_princ.encode()
+    assert new_creds.server.name == b"krbtgt/KRBTEST.COM@KRBTEST.COM"
+    # Ticket flags for new_creds should have TKT_FLG_POSTDATED set and TKT_FLG_INVALID cleared
+
+    new_creds = krb5.get_validated_creds(ctx, creds.client, cc, b"krbtgt/KRBTEST.COM@KRBTEST.COM")
+    assert new_creds.client.name == realm.user_princ.encode()
+    assert new_creds.server.name == b"krbtgt/KRBTEST.COM@KRBTEST.COM"
+    # Ticket flags for new_creds should have TKT_FLG_POSTDATED set and TKT_FLG_INVALID cleared