#!/usr/bin/env python3 # ============================================================================= # lib/krb5-load.py - a small Kerberos load generator (client simulator) # ============================================================================= # # test-kerberos.sh and start-kerberos.sh use this program. It pretends to be # many Kerberos clients that log in and ask for service tickets, and it times # every answer of the KDC. # # It uses the Kerberos library of the system (libkrb5.so.3, package # krb5-libs) through Python's built-in "ctypes" module. So every request is # exactly what kinit, sssd or a web browser would send: real # pre-authentication, real encryption, real UDP/TCP handling. It needs only # the Python 3 standard library, so it works on an offline RHEL 9 host. # # It reads the same environment as kinit: KRB5_CONFIG (which KDC, which # realm) must point to the kit's client configuration. # # MODES # probe Try to log in every 10 ms until it works. Prints the moment of # the first success. Used to measure the startup time. # as Logins only ("AS requests"). Each login returns a ticket-granting # ticket (TGT), like "kinit -k". # tgs Service tickets only ("TGS requests"). Each client logs in once # (not measured), then asks for service tickets again and again. # mix Realistic mix: a user logs in (AS), then asks for # --tgs-per-login service tickets (TGS), then the next user ... # # SPEED # --clients N N client processes run at the same time. Each one sends a # request, waits for the answer, then sends the next. # --rate R Optional: a fixed total rate (R per second, spread over the # clients) instead of "as fast as possible". # as/tgs: requests per second; mix: logins per second. # # OUTPUT (--output FILE) # Plain "key = value" lines that are easy to read and easy to parse. # With --percentiles FILE, the full latency percentile table is saved as CSV. # ============================================================================= import argparse import ctypes import math import multiprocessing import resource import sys import time from collections import Counter from ctypes import POINTER, Structure, byref, c_char_p, c_int32, c_uint, c_void_p # ============================================================================= # PART 1 - the Kerberos library (libkrb5), used through ctypes # ============================================================================= # Only the few C structures and functions this program needs are described # here. Their layout comes from /usr/include/krb5/krb5.h. class KrbData(Structure): # krb5_data: a piece of binary data _fields_ = [("magic", c_int32), ("length", c_uint), ("data", c_void_p)] class KeyBlock(Structure): # krb5_keyblock: an encryption key _fields_ = [("magic", c_int32), ("enctype", c_int32), ("length", c_uint), ("contents", c_void_p)] class TicketTimes(Structure): # krb5_ticket_times _fields_ = [("authtime", c_int32), ("starttime", c_int32), ("endtime", c_int32), ("renew_till", c_int32)] class KeytabEntry(Structure): # krb5_keytab_entry: one key of one principal _fields_ = [("magic", c_int32), ("principal", c_void_p), ("timestamp", c_int32), ("vno", c_uint), ("key", KeyBlock)] class Credentials(Structure): # krb5_creds: one ticket plus its session key _fields_ = [("magic", c_int32), ("client", c_void_p), # who the ticket is for (a krb5_principal) ("server", c_void_p), # which service it is for ("keyblock", KeyBlock), ("times", TicketTimes), ("is_skey", c_uint), ("ticket_flags", c_int32), ("addresses", c_void_p), ("ticket", KrbData), ("second_ticket", KrbData), ("authdata", c_void_p)] # krb5_get_credentials() option: do not keep the service ticket in the ticket # cache. Then every request really goes to the KDC (nothing is re-used). KRB5_GC_NO_STORE = 8 class KerberosError(Exception): pass class Kerberos: """A thin, readable wrapper around the libkrb5 functions this program uses.""" def __init__(self): lib = ctypes.CDLL("libkrb5.so.3") self.lib = lib # Describe argument and return types, so ctypes passes them correctly. # Every function returns an error code: 0 = success. def declare(name, *argument_types): function = getattr(lib, name) function.argtypes = list(argument_types) function.restype = c_int32 declare("krb5_init_context", POINTER(c_void_p)) declare("krb5_parse_name", c_void_p, c_char_p, POINTER(c_void_p)) declare("krb5_kt_resolve", c_void_p, c_char_p, POINTER(c_void_p)) declare("krb5_unparse_name", c_void_p, c_void_p, POINTER(c_void_p)) declare("krb5_kt_start_seq_get", c_void_p, c_void_p, POINTER(c_void_p)) declare("krb5_kt_next_entry", c_void_p, c_void_p, POINTER(KeytabEntry), POINTER(c_void_p)) declare("krb5_kt_end_seq_get", c_void_p, c_void_p, POINTER(c_void_p)) declare("krb5_kt_add_entry", c_void_p, c_void_p, POINTER(KeytabEntry)) declare("krb5_free_keytab_entry_contents", c_void_p, POINTER(KeytabEntry)) lib.krb5_free_unparsed_name.argtypes = [c_void_p, c_void_p] lib.krb5_free_unparsed_name.restype = None declare("krb5_get_init_creds_keytab", c_void_p, POINTER(Credentials), c_void_p, c_void_p, c_int32, c_char_p, c_void_p) declare("krb5_cc_resolve", c_void_p, c_char_p, POINTER(c_void_p)) declare("krb5_cc_initialize", c_void_p, c_void_p, c_void_p) declare("krb5_cc_store_cred", c_void_p, c_void_p, POINTER(Credentials)) declare("krb5_get_credentials", c_void_p, c_int32, c_void_p, POINTER(Credentials), POINTER(POINTER(Credentials))) lib.krb5_free_cred_contents.argtypes = [c_void_p, POINTER(Credentials)] lib.krb5_free_cred_contents.restype = None lib.krb5_free_creds.argtypes = [c_void_p, POINTER(Credentials)] lib.krb5_free_creds.restype = None lib.krb5_get_error_message.argtypes = [c_void_p, c_int32] lib.krb5_get_error_message.restype = c_void_p lib.krb5_free_error_message.argtypes = [c_void_p, c_void_p] lib.krb5_free_error_message.restype = None self.context = c_void_p() if lib.krb5_init_context(byref(self.context)) != 0: raise KerberosError("cannot initialise the Kerberos library") def _check(self, code): """Raise KerberosError with the library's own message if code != 0.""" if code != 0: pointer = self.lib.krb5_get_error_message(self.context, code) message = ctypes.string_at(pointer).decode(errors="replace") self.lib.krb5_free_error_message(self.context, pointer) raise KerberosError(message) def principal(self, name): handle = c_void_p() self._check(self.lib.krb5_parse_name(self.context, name.encode(), byref(handle))) return handle def keytab(self, name): handle = c_void_p() self._check(self.lib.krb5_kt_resolve(self.context, name.encode(), byref(handle))) return handle def load_keys(self, keytab_file, wanted_names, name_prefix): """Copy the keys of the wanted principals from a keytab file into small in-memory keytabs, one per principal. Returns {name: keytab}. Why: the library searches a keytab file from the start for every login. With thousands of test users in one file, that search would cost more time than the login itself. Real clients (sssd, services) have only a few keys in their keytab, so one small keytab per user is the realistic case.""" source = self.keytab(f"FILE:{keytab_file}") keytabs = {} cursor = c_void_p() self._check(self.lib.krb5_kt_start_seq_get(self.context, source, byref(cursor))) entry = KeytabEntry() while self.lib.krb5_kt_next_entry(self.context, source, byref(entry), byref(cursor)) == 0: text = c_void_p() self._check(self.lib.krb5_unparse_name(self.context, entry.principal, byref(text))) name = ctypes.string_at(text).decode() self.lib.krb5_free_unparsed_name(self.context, text) if name in wanted_names: if name not in keytabs: keytabs[name] = self.keytab(f"MEMORY:{name_prefix}-{len(keytabs)}") self._check(self.lib.krb5_kt_add_entry(self.context, keytabs[name], byref(entry))) self.lib.krb5_free_keytab_entry_contents(self.context, byref(entry)) self.lib.krb5_kt_end_seq_get(self.context, source, byref(cursor)) missing = set(wanted_names) - set(keytabs) if missing: raise KerberosError(f"{len(missing)} users have no key in {keytab_file}, e.g. {min(missing)}") return keytabs def login(self, user, keytab): """AS request: log in with the key from the keytab, like "kinit -k". Returns the ticket-granting ticket (free it with free()).""" tgt = Credentials() self._check(self.lib.krb5_get_init_creds_keytab( self.context, byref(tgt), user, keytab, 0, None, None)) return tgt def memory_cache(self, name, user, tgt): """A ticket cache in memory that holds this user's TGT.""" cache = c_void_p() self._check(self.lib.krb5_cc_resolve(self.context, f"MEMORY:{name}".encode(), byref(cache))) self._check(self.lib.krb5_cc_initialize(self.context, cache, user)) self._check(self.lib.krb5_cc_store_cred(self.context, cache, byref(tgt))) return cache def service_ticket(self, cache, user, service): """TGS request: ask the KDC for a ticket to one service, using the TGT in the cache. The ticket is thrown away (not stored) afterwards.""" wanted = Credentials() wanted.client = user wanted.server = service ticket = POINTER(Credentials)() self._check(self.lib.krb5_get_credentials( self.context, KRB5_GC_NO_STORE, cache, byref(wanted), byref(ticket))) self.lib.krb5_free_creds(self.context, ticket) def free(self, credentials): self.lib.krb5_free_cred_contents(self.context, byref(credentials)) # Names of the test principals (created by start-kerberos.sh). def user_name(number, realm): return f"perfuser{number:05d}@{realm}" def service_name(number, realm): return f"HTTP/web{number:04d}.perf.test@{realm}" # ============================================================================= # PART 2 - one client process # ============================================================================= # Every client process runs this function. It prepares itself, waits until # all clients are ready (the "barrier"), sends requests until the end time, # and finally sends its measurements back to the main process. def client_process(index, args, barrier, results_pipe): krb = Kerberos() # This client's share of the test users: client 0 uses users 1, 1+N, # 1+2N ..., client 1 uses users 2, 2+N ... (N = number of clients). my_user_numbers = list(range(index + 1, args.users + 1, args.clients)) or [index % args.users + 1] user_names = [user_name(n, args.realm) for n in my_user_numbers] keytabs = krb.load_keys(args.keytab, set(user_names), f"client{index}") users = [(krb.principal(name), keytabs[name]) for name in user_names] services = [krb.principal(service_name(n, args.realm)) for n in range(1, args.services + 1)] # For the "tgs" mode: log in once now (not measured). cache = None if args.mode == "tgs": first_user, first_keytab = users[0] tgt = krb.login(first_user, first_keytab) cache = krb.memory_cache(f"client{index}", first_user, tgt) krb.free(tgt) as_times_ms, tgs_times_ms = [], [] errors = Counter() units_late = 0 # rate mode: started more than 100 ms behind schedule def timed(times, action, *arguments): """Run one request, note its duration (ms) or its error. Returns the result or None.""" start = time.perf_counter() try: result = action(*arguments) except KerberosError as error: errors[str(error)] += 1 return None times.append((time.perf_counter() - start) * 1000.0) return result # --- wait until every client is ready, so they all start together -------- barrier.wait() cpu_start = resource.getrusage(resource.RUSAGE_SELF) start = time.monotonic() end = start + args.duration # With a fixed rate, this client starts one "unit" (a request, or in mix # mode a whole login session) every "interval" seconds. The clients are # spread out a little, so they do not all fire at the same moment. interval = args.clients / args.rate if args.rate else 0.0 next_start = start + interval * index / args.clients unit = 0 while True: now = time.monotonic() if now >= end: break if interval: if next_start >= end: break if now < next_start: time.sleep(next_start - now) elif now - next_start > 0.1: units_late += 1 next_start += interval user, keytab = users[unit % len(users)] if args.mode == "as": tgt = timed(as_times_ms, krb.login, user, keytab) if tgt is not None: krb.free(tgt) elif args.mode == "tgs": service = services[(index + unit) % len(services)] timed(tgs_times_ms, krb.service_ticket, cache, users[0][0], service) elif args.mode == "mix": tgt = timed(as_times_ms, krb.login, user, keytab) if tgt is not None: session_cache = krb.memory_cache(f"client{index}", user, tgt) krb.free(tgt) for k in range(args.tgs_per_login): service = services[(index + unit * args.tgs_per_login + k) % len(services)] timed(tgs_times_ms, krb.service_ticket, session_cache, user, service) unit += 1 elapsed = time.monotonic() - start cpu_end = resource.getrusage(resource.RUSAGE_SELF) cpu_seconds = (cpu_end.ru_utime - cpu_start.ru_utime) + (cpu_end.ru_stime - cpu_start.ru_stime) results_pipe.send({ "as_times_ms": as_times_ms, "tgs_times_ms": tgs_times_ms, "errors": dict(errors), "elapsed": elapsed, "cpu_seconds": cpu_seconds, "units": unit, "units_late": units_late, }) results_pipe.close() # ============================================================================= # PART 3 - statistics and output # ============================================================================= def percentile(sorted_values, pct): """The value below which pct percent of the measurements fall.""" if not sorted_values: return 0.0 position = max(0, math.ceil(pct / 100.0 * len(sorted_values)) - 1) return sorted_values[min(position, len(sorted_values) - 1)] def latency_lines(prefix, times_ms): """Average, percentiles and maximum as "key = value" lines.""" values = sorted(times_ms) lines = [] average = sum(values) / len(values) if values else 0.0 lines.append(f"{prefix}_avg_ms = {average:.3f}") for pct in (50, 90, 95, 99, 99.9): name = str(pct).replace(".", "_") lines.append(f"{prefix}_p{name}_ms = {percentile(values, pct):.3f}") lines.append(f"{prefix}_max_ms = {values[-1] if values else 0.0:.3f}") return lines def write_percentile_table(path, as_times_ms, tgs_times_ms): as_sorted, tgs_sorted = sorted(as_times_ms), sorted(tgs_times_ms) with open(path, "w") as table: table.write("percentile,as_login_ms,tgs_service_ticket_ms\n") for pct in range(1, 101): table.write(f"{pct},{percentile(as_sorted, pct):.3f},{percentile(tgs_sorted, pct):.3f}\n") def run_load(args): """Start the client processes, collect their results, write the summary.""" context = multiprocessing.get_context("fork") barrier = context.Barrier(args.clients + 1) pipes, processes = [], [] for index in range(args.clients): receiver, sender = context.Pipe(duplex=False) process = context.Process(target=client_process, args=(index, args, barrier, sender)) process.start() pipes.append(receiver) processes.append(process) # Wait until every client has prepared itself (at most 60 s), then go. try: barrier.wait(timeout=60) except multiprocessing.BrokenBarrierError: for process in processes: process.terminate() sys.exit("A client process could not prepare itself (is the KDC running?)") results = [pipe.recv() for pipe in pipes] for process in processes: process.join() as_times = [t for r in results for t in r["as_times_ms"]] tgs_times = [t for r in results for t in r["tgs_times_ms"]] errors = Counter() for r in results: errors.update(r["errors"]) elapsed = max(r["elapsed"] for r in results) cpu_seconds = sum(r["cpu_seconds"] for r in results) units = sum(r["units"] for r in results) units_late = sum(r["units_late"] for r in results) requests_ok = len(as_times) + len(tgs_times) requests_failed = sum(errors.values()) requests_total = requests_ok + requests_failed failed_pct = 100.0 * requests_failed / requests_total if requests_total else 0.0 lines = [ "# Kerberos load generator results (lib/krb5-load.py)", f"mode = {args.mode}", f"clients = {args.clients}", f"target_rate_per_s = {args.rate or 0}", f"tgs_per_login = {args.tgs_per_login if args.mode == 'mix' else 0}", f"duration_s = {elapsed:.3f}", "", "# Requests: an AS request is a login, a TGS request asks for a service ticket", f"requests_ok = {requests_ok}", f"requests_failed = {requests_failed}", f"requests_total = {requests_total}", f"failed_pct = {failed_pct:.4f}", f"rate_per_s = {requests_ok / elapsed:.3f}", f"as_ok = {len(as_times)}", f"as_per_s = {len(as_times) / elapsed:.3f}", f"tgs_ok = {len(tgs_times)}", f"tgs_per_s = {len(tgs_times) / elapsed:.3f}", f"units_started = {units}", f"units_late_pct = {100.0 * units_late / units if units else 0.0:.3f}", "", "# Latency in milliseconds (time from sending the request until the ticket is ready)", ] lines += latency_lines("as", as_times) lines += latency_lines("tgs", tgs_times) lines += [ "", "# How busy the client processes were (100% = they never waited for the KDC)", f"generator_cpu_seconds = {cpu_seconds:.3f}", f"generator_busy_pct = {100.0 * cpu_seconds / (args.clients * elapsed):.1f}", "", "# Errors, most frequent first: error_count_N = how often, error_text_N = message", f"error_kinds = {len(errors)}", ] for number, (message, count) in enumerate(errors.most_common(10), start=1): lines.append(f"error_count_{number} = {count}") lines.append(f"error_text_{number} = {message}") output = "\n".join(lines) + "\n" if args.output: with open(args.output, "w") as out_file: out_file.write(output) else: sys.stdout.write(output) if args.percentiles: write_percentile_table(args.percentiles, as_times, tgs_times) # ============================================================================= # PART 4 - the "probe" mode (startup time) # ============================================================================= def run_probe(args): """Try to log in every 10 ms, until it works or --timeout seconds pass.""" krb = Kerberos() name = user_name(1, args.realm) keytab = krb.load_keys(args.keytab, {name}, "probe")[name] user = krb.principal(name) # Tell the caller we are ready: from now on every attempt is timed. if args.ready_file: open(args.ready_file, "w").close() deadline = time.monotonic() + args.timeout attempts = 0 while time.monotonic() < deadline: attempts += 1 try: tgt = krb.login(user, keytab) except KerberosError: time.sleep(0.01) continue success_time = time.time() krb.free(tgt) output = f"first_success_epoch = {success_time:.6f}\nattempts = {attempts}\n" if args.output: with open(args.output, "w") as out_file: out_file.write(output) else: sys.stdout.write(output) return 0 return 1 # ============================================================================= # MAIN # ============================================================================= def main(): parser = argparse.ArgumentParser(description="Kerberos KDC load generator (see comments in this file)") parser.add_argument("mode", choices=["probe", "as", "tgs", "mix"]) parser.add_argument("--realm", required=True) parser.add_argument("--keytab", required=True, help="keytab with the test users' keys") parser.add_argument("--users", type=int, default=1000, help="number of test users") parser.add_argument("--services", type=int, default=100, help="number of test services") parser.add_argument("--clients", type=int, default=8, help="client processes at the same time") parser.add_argument("--duration", type=float, default=10, help="seconds of load") parser.add_argument("--rate", type=float, default=0, help="fixed total rate per second (0 = full speed)") parser.add_argument("--tgs-per-login", type=int, default=4, help="mix mode: service tickets per login") parser.add_argument("--timeout", type=float, default=60, help="probe mode: give up after this many seconds") parser.add_argument("--ready-file", help="probe mode: create this file when ready") parser.add_argument("--output", help="write the results to this file (default: screen)") parser.add_argument("--percentiles", help="write latency percentiles 1..100 to this CSV file") args = parser.parse_args() if args.mode == "probe": return run_probe(args) run_load(args) return 0 if __name__ == "__main__": sys.exit(main())