85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
import os
|
|
import shutil
|
|
import subprocess
|
|
from typing import List
|
|
|
|
from zope import component
|
|
from zope.interface import implementer
|
|
|
|
from ceo_common.interfaces import IKerberosService, IConfig
|
|
from ceo_common.krb5._krb5 import ffi, lib
|
|
from ceo_common.krb5.utils import check_rc, get_krb5_context, \
|
|
get_krb5_cc_default
|
|
|
|
|
|
@implementer(IKerberosService)
|
|
class KerberosService:
|
|
def __init__(
|
|
self,
|
|
admin_principal: str,
|
|
):
|
|
cfg = component.getUtility(IConfig)
|
|
|
|
self.admin_principal = admin_principal
|
|
self.cache_dir = cfg.get('ceod_krb5_cache_dir')
|
|
self.realm = cfg.get('ldap_sasl_realm')
|
|
self._initialize_cache()
|
|
|
|
def _initialize_cache(self, **kwargs):
|
|
if os.path.isdir(self.cache_dir):
|
|
shutil.rmtree(self.cache_dir)
|
|
os.makedirs(self.cache_dir)
|
|
os.environ['KRB5CCNAME'] = 'DIR:' + self.cache_dir
|
|
|
|
with get_krb5_context() as k_ctx, get_krb5_cc_default(k_ctx) as cache:
|
|
princ = None
|
|
try:
|
|
# build a principal for 'nobody'
|
|
realm = self.realm.encode()
|
|
c_realm = ffi.new('char[]', realm)
|
|
component = ffi.new('char[]', b'nobody')
|
|
p_princ = ffi.new('krb5_principal *')
|
|
rc = lib.krb5_build_principal(
|
|
k_ctx, p_princ, len(realm), c_realm, component, ffi.NULL)
|
|
check_rc(k_ctx, rc)
|
|
princ = p_princ[0]
|
|
|
|
# initialize the default cache with 'nobody' as the default principal
|
|
rc = lib.krb5_cc_initialize(k_ctx, cache, princ)
|
|
check_rc(k_ctx, rc)
|
|
finally:
|
|
if princ is not None:
|
|
lib.krb5_free_principal(k_ctx, princ)
|
|
|
|
def _run(self, args: List[str]):
|
|
subprocess.run(args, check=True)
|
|
|
|
def addprinc(self, principal: str, password: str):
|
|
self._run([
|
|
'kadmin', '-k', '-p', self.admin_principal, 'addprinc',
|
|
'-pw', password,
|
|
'-policy', 'default',
|
|
'+needchange',
|
|
'+requires_preauth',
|
|
principal
|
|
])
|
|
|
|
def delprinc(self, principal: str):
|
|
self._run([
|
|
'kadmin', '-k', '-p', self.admin_principal, 'delprinc',
|
|
'-force',
|
|
principal
|
|
])
|
|
|
|
def change_password(self, principal: str, password: str):
|
|
self._run([
|
|
'kadmin', '-k', '-p', self.admin_principal, 'cpw',
|
|
'-pw', password,
|
|
principal
|
|
])
|
|
self._run([
|
|
'kadmin', '-k', '-p', self.admin_principal, 'modprinc',
|
|
'+needchange',
|
|
principal
|
|
])
|