202 lines
6.7 KiB

import contextlib
import functools
from typing import Union
from ._krb5 import ffi, lib
from ceo_common.errors import KerberosError
def check_rc(k_ctx, rc: int):
Check the return code of a krb5 function.
An exception is raised if rc != 0.
if rc == 0:
c_msg = lib.krb5_get_error_message(k_ctx, rc)
msg = ffi.string(c_msg).decode()
lib.krb5_free_error_message(k_ctx, c_msg)
raise KerberosError(msg)
def get_krb5_context():
"""Yields a krb5_context."""
k_ctx = None
p_k_ctx ='krb5_context *')
rc = lib.krb5_init_context(p_k_ctx)
k_ctx = p_k_ctx[0] # dereference the pointer
check_rc(k_ctx, rc)
yield k_ctx
if k_ctx is not None:
def get_krb5_auth_context(k_ctx):
"""Yields a krb5_auth_context."""
a_ctx = None
p_a_ctx ='krb5_auth_context *')
rc = lib.krb5_auth_con_init(k_ctx, p_a_ctx)
a_ctx = p_a_ctx[0]
check_rc(k_ctx, rc)
# clear the flags which enable the replay cache
rc = lib.krb5_auth_con_setflags(k_ctx, a_ctx, 0)
check_rc(k_ctx, rc)
yield a_ctx
if a_ctx is not None:
lib.krb5_auth_con_free(k_ctx, a_ctx)
def get_krb5_cc_default(k_ctx):
"""Yields the default krb5_ccache."""
cache = None
p_cache ='krb5_ccache *')
rc = lib.krb5_cc_default(k_ctx, p_cache)
check_rc(k_ctx, rc)
cache = p_cache[0]
yield cache
if cache is not None:
lib.krb5_cc_close(k_ctx, cache)
def get_krb5_cc_resolve(k_ctx, name: str):
Resolve a credential cache name.
`name` should have the format 'type:residual'.
cache = None
c_name ='char[]', name.encode())
p_cache ='krb5_ccache *')
rc = lib.krb5_cc_resolve(k_ctx, c_name, p_cache)
check_rc(k_ctx, rc)
cache = p_cache[0]
yield cache
if cache is not None:
lib.krb5_cc_close(k_ctx, cache)
def get_fwd_tgt(server: str, cache_name: Union[str, None] = None) -> bytes:
Get a forwarded TGT formatted as a KRB-CRED message.
`server` should have the format 'service/host',
e.g. 'ceod/'.
If `cache_name` is None, the default cache will be used; otherwise,
the cache with that name will be used.
if cache_name is None:
cache_function = get_krb5_cc_default
cache_function = functools.partial(get_krb5_cc_resolve, name=cache_name)
with get_krb5_context() as k_ctx, \
get_krb5_auth_context(k_ctx) as a_ctx, \
cache_function(k_ctx) as cache:
server_princ = None
client_princ = None
outbuf = None
# create a server principal from the server string
c_server ='char[]', server.encode())
p_server_princ ='krb5_principal *')
rc = lib.krb5_parse_name(k_ctx, c_server, p_server_princ)
check_rc(k_ctx, rc)
server_princ = p_server_princ[0]
# get a client principal from the default cache
p_client_princ ='krb5_principal *')
rc = lib.krb5_cc_get_principal(k_ctx, cache, p_client_princ)
check_rc(k_ctx, rc)
client_princ = p_client_princ[0]
# get the forwarded TGT
p_outbuf ='krb5_data *')
rc = lib.krb5_fwd_tgt_creds(
k_ctx, a_ctx, ffi.NULL, client_princ, server_princ,
cache, 0, p_outbuf)
check_rc(k_ctx, rc)
outbuf = p_outbuf[0]
return ffi.unpack(, outbuf.length)
if outbuf is not None:
lib.krb5_free_data_contents(k_ctx, p_outbuf)
if client_princ is not None:
lib.krb5_free_principal(k_ctx, client_princ)
if server_princ is not None:
lib.krb5_free_principal(k_ctx, server_princ)
def store_fwd_tgt_creds(cred_data_bytes: bytes):
Stores the credentials found in cred_data_bytes
in a new sub-cache in the default cache (which should be of type DIR),
yields the name of the principal in the credential, then finally removes
the sub-cache from the collection.
The principal name will have the form 'username@REALM', e.g.
with get_krb5_context() as k_ctx, get_krb5_auth_context(k_ctx) as a_ctx:
creds_out = None
cache = None
c_name = None
# fill a krb5_data struct
cred_data ='krb5_data *')
cred_data_buf ='char[]', cred_data_bytes)
cred_data.magic = lib.KV5M_DATA
cred_data.length = len(cred_data_bytes) = cred_data_buf
# read the KRB5-CRED message into a krb5_creds array
p_creds_out ='krb5_creds ***')
rc = lib.krb5_rd_cred(k_ctx, a_ctx, cred_data, p_creds_out, ffi.NULL)
check_rc(k_ctx, rc)
creds_out = p_creds_out[0]
# there should only be one cred in the array
cred = creds_out[0]
# read the name of the client principal
client_princ = cred.client
p_name ='char **')
rc = lib.krb5_unparse_name(k_ctx, client_princ, p_name)
check_rc(k_ctx, rc)
c_name = p_name[0]
name = ffi.string(c_name).decode()
# create a new cache inside the collection (default cache)
p_cache ='krb5_ccache *')
cctype ='char[]', b'DIR')
rc = lib.krb5_cc_new_unique(k_ctx, cctype, ffi.NULL, p_cache)
check_rc(k_ctx, rc)
cache = p_cache[0]
# initialize the new cache
rc = lib.krb5_cc_initialize(k_ctx, cache, client_princ)
check_rc(k_ctx, rc)
# store the cred into the cache
rc = lib.krb5_cc_store_cred(k_ctx, cache, cred)
check_rc(k_ctx, rc)
yield name
if cache is not None:
# We destroy the cache (instead of closing it) since we want
# to remove it from disk.
lib.krb5_cc_destroy(k_ctx, cache)
if c_name is not None:
lib.krb5_free_unparsed_name(k_ctx, c_name)
if creds_out is not None:
lib.krb5_free_tgt_creds(k_ctx, creds_out)