Add krb5_cc_retrieve_cred() and krb5_cc_remove_cred()

src/krb5/__init__.py

@@ -3,6 +3,7 @@
 
 from krb5._ccache import (
     CCache,
+    CredentialsRetrieveFlags,
     cc_default,
     cc_default_name,
     cc_destroy,
@@ -11,7 +12,9 @@
     cc_get_type,
     cc_initialize,
     cc_new_unique,
+    cc_remove_cred,
     cc_resolve,
+    cc_retrieve_cred,
     cc_set_default_name,
     cc_store_cred,
     cc_switch,
@@ -70,6 +73,7 @@
 __all__ = [
     "CCache",
     "Context",
+    "CredentialsRetrieveFlags",
     "Creds",
     "GetInitCredsOpt",
     "InitCredsContext",
@@ -89,7 +93,9 @@
     "cc_get_type",
     "cc_initialize",
     "cc_new_unique",
+    "cc_remove_cred",
     "cc_resolve",
+    "cc_retrieve_cred",
     "cc_set_default_name",
     "cc_store_cred",
     "cc_switch",

src/krb5/_ccache.pyi

@@ -1,12 +1,28 @@
 # Copyright: (c) 2021 Jordan Borean (@jborean93) <jborean93@gmail.com>
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 
+import enum
 import typing
 
 from krb5._context import Context
 from krb5._creds import Creds
 from krb5._principal import Principal
 
+class CredentialsRetrieveFlags(enum.IntEnum):
+    """Flags used to control :meth:`cc_retrieve_cred` and  :meth:`cc_remove_cred`."""
+
+    none: CredentialsRetrieveFlags = ...  #: No matching flags set
+    match_times: CredentialsRetrieveFlags = ...  #: The requested lifetime must be at least as great as specified
+    match_is_skey: CredentialsRetrieveFlags = ...  #: The is_skey field must match exactly
+    match_flags: CredentialsRetrieveFlags = ...  #: All the flags set in the match credentials must be set
+    match_times_exact: CredentialsRetrieveFlags = ...  #: All the time fields must match exactly
+    match_flags_exact: CredentialsRetrieveFlags = ...  #: All the flags must match exactly
+    match_authdata: CredentialsRetrieveFlags = ...  #: The authorization data must match
+    match_srv_nameonly: CredentialsRetrieveFlags = ...  #: Only the name portion of the principal name must match
+    match_2nd_tkt: CredentialsRetrieveFlags = ...  #: The second ticket must match
+    match_keytype: CredentialsRetrieveFlags = ...  #: The encryption key type must match
+    supported_ktypes: CredentialsRetrieveFlags = ...  #: The supported key types must match
+
 class CCache:
     """Kerberos CCache
 
@@ -158,6 +174,24 @@ def cc_new_unique(
         CCache: The created credential cache.
     """
 
+def cc_remove_cred(
+    context: Context,
+    cache: CCache,
+    flags: typing.Union[int, CredentialsRetrieveFlags],
+    creds: Creds,
+) -> None:
+    """Remove matching credentials from a credential cache.
+
+    Remove all credential which match creds according to flags from the
+    credential cache.
+
+    Args:
+        context: Krb5 context.
+        cache: The credential cache to store the creds into.
+        flags: The flags describing how to perform the matching.
+        creds: The credentials to match against.
+    """
+
 def cc_resolve(
     context: Context,
     name: bytes,
@@ -177,6 +211,27 @@ def cc_resolve(
         CCache: The credential cache that was resolved.
     """
 
+def cc_retrieve_cred(
+    context: Context,
+    cache: CCache,
+    flags: typing.Union[int, CredentialsRetrieveFlags],
+    mcreds: Creds,
+) -> Creds:
+    """Retrieve matching credentials from a credential cache.
+
+    Retrieve all credential which match creds according to flags from the
+    credential cache.
+
+    Args:
+        context: Krb5 context.
+        cache: The credential cache to store the creds into.
+        flags: The flags describing how to perform the matching.
+        mcreds: The credentials to match against.
+
+    Returns:
+        Creds: The matching credentials.
+    """
+
 def cc_set_default_name(
     context: Context,
     name: typing.Optional[bytes],

src/krb5/_ccache.pyx

@@ -1,6 +1,7 @@
 # Copyright: (c) 2021 Jordan Borean (@jborean93) <jborean93@gmail.com>
 # MIT License (see LICENSE or https://opensource.org/licenses/MIT)
 
+import enum
 import typing
 
 from libc.stdint cimport uintptr_t
@@ -76,12 +77,27 @@ cdef extern from "python_krb5.h":
         krb5_creds *cred,
     ) nogil
 
+    krb5_error_code krb5_cc_remove_cred(
+        krb5_context context,
+        krb5_ccache cache,
+        int flags,
+        krb5_creds *creds,
+    ) nogil
+
     krb5_error_code krb5_cc_resolve(
         krb5_context context,
         const char *name,
         krb5_ccache *cache,
     ) nogil
 
+    krb5_error_code krb5_cc_retrieve_cred(
+        krb5_context context,
+        krb5_ccache cache,
+        int flags,
+        krb5_creds *mcreds,
+        krb5_creds *creds,
+    ) nogil
+
     krb5_error_code krb5_cc_set_default_name(
         krb5_context context,
         const char *name,
@@ -104,6 +120,38 @@ cdef extern from "python_krb5.h":
         krb5_ccache cache,
     ) nogil
 
+    int32_t KRB5_TC_MATCH_TIMES
+    int32_t KRB5_TC_MATCH_IS_SKEY
+    int32_t KRB5_TC_MATCH_FLAGS
+    int32_t KRB5_TC_MATCH_TIMES_EXACT
+    int32_t KRB5_TC_MATCH_FLAGS_EXACT
+    int32_t KRB5_TC_MATCH_AUTHDATA
+    int32_t KRB5_TC_MATCH_SRV_NAMEONLY
+    int32_t KRB5_TC_MATCH_2ND_TKT
+    int32_t KRB5_TC_MATCH_KTYPE
+    int32_t KRB5_TC_SUPPORTED_KTYPES
+
+
+_CredentialsRetrieveFlags_members = [
+    ('none', 0),
+    ('match_times', KRB5_TC_MATCH_TIMES),
+    ('match_is_skey', KRB5_TC_MATCH_IS_SKEY),
+    ('match_flags', KRB5_TC_MATCH_FLAGS),
+    ('match_times_exact', KRB5_TC_MATCH_TIMES_EXACT),
+    ('match_flags_exact', KRB5_TC_MATCH_FLAGS_EXACT),
+    ('match_authdata', KRB5_TC_MATCH_AUTHDATA),
+    ('match_srv_nameonly', KRB5_TC_MATCH_SRV_NAMEONLY),
+    ('match_2nd_tkt', KRB5_TC_MATCH_2ND_TKT),
+    ('match_keytype', KRB5_TC_MATCH_KTYPE),
+]
+# If KRB5_TC_SUPPORTED_KTYPES is not available it will be set to 0 in
+# python_krb5.h
+if KRB5_TC_SUPPORTED_KTYPES != 0:
+    _CredentialsRetrieveFlags_members += [
+        ('supported_ktypes', KRB5_TC_SUPPORTED_KTYPES),
+    ]
+CredentialsRetrieveFlags = enum.IntEnum('CredentialsRetrieveFlags', _CredentialsRetrieveFlags_members)
+
 
 cdef class CCache:
     # cdef Context ctx
@@ -278,6 +326,19 @@ def cc_new_unique(
     return ccache
 
 
+def cc_remove_cred(
+    Context context not None,
+    CCache cache not None,
+    int flags,
+    Creds creds not None,
+) -> None:
+    cdef krb5_error_code err = 0
+
+    err = krb5_cc_remove_cred(context.raw, cache.raw, flags, &creds.raw)
+    if err:
+        raise Krb5Error(context, err)
+
+
 def cc_resolve(
     Context context not None,
     const unsigned char[:] name not None,
@@ -298,6 +359,22 @@ def cc_resolve(
     return ccache
 
 
+def cc_retrieve_cred(
+    Context context not None,
+    CCache cache not None,
+    int flags,
+    Creds mcreds not None,
+) -> Creds:
+    creds = Creds(context)
+    cdef krb5_error_code err = 0
+
+    err = krb5_cc_retrieve_cred(context.raw, cache.raw, flags, &mcreds.raw, &creds.raw)
+    if err:
+        raise Krb5Error(context, err)
+
+    return creds
+
+
 def cc_set_default_name(
     Context context not None,
     const unsigned char[:] name,

src/krb5/python_krb5.h

@@ -9,3 +9,8 @@
 #ifndef KRB5_KT_PREFIX_MAX_LEN
 #define KRB5_KT_PREFIX_MAX_LEN -1
 #endif
+
+// Heimdal does not define this
+#ifndef KRB5_TC_SUPPORTED_KTYPES
+#define KRB5_TC_SUPPORTED_KTYPES 0
+#endif

tests/test_ccache.py

@@ -4,6 +4,7 @@
 import os
 import os.path
 import pathlib
+import platform
 import sys
 
 import k5test
@@ -264,3 +265,32 @@ def test_cc_cache_match(realm: k5test.K5Realm, tmp_path: pathlib.Path, monkeypat
     assert user_actual.name == b":" + bytes(tmp_path) + b"/tkt-user"
     assert user_actual.principal
     assert user_actual.principal.name == user_princ.name
+
+
+def test_cc_retrieve_remove_cred(realm: k5test.K5Realm, tmp_path: pathlib.Path) -> 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())
+
+    cc = krb5.cc_resolve(ctx, f"{tmp_path / 'ccache'}".encode())
+    krb5.cc_initialize(ctx, cc, princ)
+
+    msg_pattern = "Matching credential not found|End of credential cache reached|Did not find credential for"
+    with pytest.raises(krb5.Krb5Error, match=msg_pattern):
+        c = krb5.cc_retrieve_cred(ctx, cc, krb5.CredentialsRetrieveFlags.match_srv_nameonly, creds)
+
+    assert len(list(cc)) == 0
+    krb5.cc_store_cred(ctx, cc, creds)
+    assert len(list(cc)) > 0
+
+    krb5.cc_retrieve_cred(ctx, cc, krb5.CredentialsRetrieveFlags.match_srv_nameonly, creds)
+
+    krb5.cc_remove_cred(ctx, cc, krb5.CredentialsRetrieveFlags.match_srv_nameonly, creds)
+
+    if realm.provider.lower() == "heimdal" and platform.system() == "Linux":
+        pytest.skip("Removing credentials does not seem to have an effect with heimdal on Linux")
+
+    msg_pattern = "Matching credential not found|End of credential cache reached|Did not find credential for"
+    with pytest.raises(krb5.Krb5Error, match=msg_pattern):
+        krb5.cc_retrieve_cred(ctx, cc, krb5.CredentialsRetrieveFlags.match_srv_nameonly, creds)