summaryrefslogtreecommitdiffstats
path: root/ipapython
diff options
context:
space:
mode:
Diffstat (limited to 'ipapython')
-rw-r--r--ipapython/dnssec/__init__.py0
-rw-r--r--ipapython/dnssec/abshsm.py184
-rw-r--r--ipapython/dnssec/bindmgr.py217
-rw-r--r--ipapython/dnssec/keysyncer.py191
-rw-r--r--ipapython/dnssec/ldapkeydb.py450
-rwxr-xr-xipapython/dnssec/localhsm.py227
-rw-r--r--ipapython/dnssec/odsmgr.py210
-rw-r--r--ipapython/dnssec/syncrepl.py115
-rw-r--r--ipapython/dnssec/temp.py22
-rw-r--r--ipapython/p11helper.py1772
-rw-r--r--ipapython/secrets/__init__.py0
-rw-r--r--ipapython/secrets/client.py109
-rw-r--r--ipapython/secrets/common.py45
-rw-r--r--ipapython/secrets/kem.py228
-rw-r--r--ipapython/secrets/store.py261
-rwxr-xr-xipapython/setup.py10
16 files changed, 0 insertions, 4041 deletions
diff --git a/ipapython/dnssec/__init__.py b/ipapython/dnssec/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/ipapython/dnssec/__init__.py
+++ /dev/null
diff --git a/ipapython/dnssec/abshsm.py b/ipapython/dnssec/abshsm.py
deleted file mode 100644
index 1533892f8..000000000
--- a/ipapython/dnssec/abshsm.py
+++ /dev/null
@@ -1,184 +0,0 @@
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-from ipapython import p11helper as _ipap11helper
-
-attrs_id2name = {
- #_ipap11helper.CKA_ALLOWED_MECHANISMS: 'ipk11allowedmechanisms',
- _ipap11helper.CKA_ALWAYS_AUTHENTICATE: 'ipk11alwaysauthenticate',
- _ipap11helper.CKA_ALWAYS_SENSITIVE: 'ipk11alwayssensitive',
- #_ipap11helper.CKA_CHECK_VALUE: 'ipk11checkvalue',
- _ipap11helper.CKA_COPYABLE: 'ipk11copyable',
- _ipap11helper.CKA_DECRYPT: 'ipk11decrypt',
- _ipap11helper.CKA_DERIVE: 'ipk11derive',
- #_ipap11helper.CKA_DESTROYABLE: 'ipk11destroyable',
- _ipap11helper.CKA_ENCRYPT: 'ipk11encrypt',
- #_ipap11helper.CKA_END_DATE: 'ipk11enddate',
- _ipap11helper.CKA_EXTRACTABLE: 'ipk11extractable',
- _ipap11helper.CKA_ID: 'ipk11id',
- #_ipap11helper.CKA_KEY_GEN_MECHANISM: 'ipk11keygenmechanism',
- _ipap11helper.CKA_KEY_TYPE: 'ipk11keytype',
- _ipap11helper.CKA_LABEL: 'ipk11label',
- _ipap11helper.CKA_LOCAL: 'ipk11local',
- _ipap11helper.CKA_MODIFIABLE: 'ipk11modifiable',
- _ipap11helper.CKA_NEVER_EXTRACTABLE: 'ipk11neverextractable',
- _ipap11helper.CKA_PRIVATE: 'ipk11private',
- #_ipap11helper.CKA_PUBLIC_KEY_INFO: 'ipapublickey',
- #_ipap11helper.CKA_PUBLIC_KEY_INFO: 'ipk11publickeyinfo',
- _ipap11helper.CKA_SENSITIVE: 'ipk11sensitive',
- _ipap11helper.CKA_SIGN: 'ipk11sign',
- _ipap11helper.CKA_SIGN_RECOVER: 'ipk11signrecover',
- #_ipap11helper.CKA_START_DATE: 'ipk11startdate',
- #_ipap11helper.CKA_SUBJECT: 'ipk11subject',
- _ipap11helper.CKA_TRUSTED: 'ipk11trusted',
- _ipap11helper.CKA_UNWRAP: 'ipk11unwrap',
- #_ipap11helper.CKA_UNWRAP_TEMPLATE: 'ipk11unwraptemplate',
- _ipap11helper.CKA_VERIFY: 'ipk11verify',
- _ipap11helper.CKA_VERIFY_RECOVER: 'ipk11verifyrecover',
- _ipap11helper.CKA_WRAP: 'ipk11wrap',
- #_ipap11helper.CKA_WRAP_TEMPLATE: 'ipk11wraptemplate',
- _ipap11helper.CKA_WRAP_WITH_TRUSTED: 'ipk11wrapwithtrusted',
-}
-
-attrs_name2id = {v: k for k, v in attrs_id2name.items()}
-
-# attribute:
-# http://www.freeipa.org/page/V4/PKCS11_in_LDAP/Schema#ipk11KeyType
-#
-# mapping table:
-# http://www.freeipa.org/page/V4/PKCS11_in_LDAP/Schema#CK_MECHANISM_TYPE
-keytype_name2id = {
- "rsa": _ipap11helper.KEY_TYPE_RSA,
- "aes": _ipap11helper.KEY_TYPE_AES,
- }
-
-keytype_id2name = {v: k for k, v in keytype_name2id.items()}
-
-wrappingmech_name2id = {
- "rsaPkcs": _ipap11helper.MECH_RSA_PKCS,
- "rsaPkcsOaep": _ipap11helper.MECH_RSA_PKCS_OAEP,
- "aesKeyWrap": _ipap11helper.MECH_AES_KEY_WRAP,
- "aesKeyWrapPad": _ipap11helper.MECH_AES_KEY_WRAP_PAD
- }
-
-wrappingmech_id2name = {v: k for k, v in wrappingmech_name2id.items()}
-
-
-bool_attr_names = set([
- 'ipk11alwaysauthenticate',
- 'ipk11alwayssensitive',
- 'ipk11copyable',
- 'ipk11decrypt',
- 'ipk11derive',
- 'ipk11encrypt',
- 'ipk11extractable',
- 'ipk11local',
- 'ipk11modifiable',
- 'ipk11neverextractable',
- 'ipk11private',
- 'ipk11sensitive',
- 'ipk11sign',
- 'ipk11signrecover',
- 'ipk11trusted',
- 'ipk11unwrap',
- 'ipk11verify',
- 'ipk11verifyrecover',
- 'ipk11wrap',
- 'ipk11wrapwithtrusted',
-])
-
-modifiable_attrs_id2name = {
- _ipap11helper.CKA_DECRYPT: 'ipk11decrypt',
- _ipap11helper.CKA_DERIVE: 'ipk11derive',
- _ipap11helper.CKA_ENCRYPT: 'ipk11encrypt',
- _ipap11helper.CKA_EXTRACTABLE: 'ipk11extractable',
- _ipap11helper.CKA_ID: 'ipk11id',
- _ipap11helper.CKA_LABEL: 'ipk11label',
- _ipap11helper.CKA_SENSITIVE: 'ipk11sensitive',
- _ipap11helper.CKA_SIGN: 'ipk11sign',
- _ipap11helper.CKA_SIGN_RECOVER: 'ipk11signrecover',
- _ipap11helper.CKA_UNWRAP: 'ipk11unwrap',
- _ipap11helper.CKA_VERIFY: 'ipk11verify',
- _ipap11helper.CKA_VERIFY_RECOVER: 'ipk11verifyrecover',
- _ipap11helper.CKA_WRAP: 'ipk11wrap',
-}
-
-modifiable_attrs_name2id = {v: k for k, v in modifiable_attrs_id2name.items()}
-
-def sync_pkcs11_metadata(log, source, target):
- """sync ipk11 metadata from source object to target object"""
-
- # iterate over list of modifiable PKCS#11 attributes - this prevents us
- # from attempting to set read-only attributes like CKA_LOCAL
- for attr in modifiable_attrs_name2id:
- if attr in source:
- if source[attr] != target[attr]:
- log.debug('Updating attribute %s from "%s" to "%s"', attr, repr(source[attr]), repr(target[attr]))
- target[attr] = source[attr]
-
-def populate_pkcs11_metadata(source, target):
- """populate all ipk11 metadata attributes in target object from source object"""
- for attr in attrs_name2id:
- if attr in source:
- target[attr] = source[attr]
-
-def ldap2p11helper_api_params(ldap_key):
- """prepare dict with metadata parameters suitable for key unwrapping"""
- unwrap_params = {}
-
- # some attributes are just renamed
- direct_param_map = {
- "ipk11label": "label",
- "ipk11id": "id",
- "ipk11copyable": "cka_copyable",
- "ipk11decrypt": "cka_decrypt",
- "ipk11derive": "cka_derive",
- "ipk11encrypt": "cka_encrypt",
- "ipk11extractable": "cka_extractable",
- "ipk11modifiable": "cka_modifiable",
- "ipk11private": "cka_private",
- "ipk11sensitive": "cka_sensitive",
- "ipk11sign": "cka_sign",
- "ipk11unwrap": "cka_unwrap",
- "ipk11verify": "cka_verify",
- "ipk11wrap": "cka_wrap",
- "ipk11wrapwithtrusted": "cka_wrap_with_trusted"
- }
-
- for ldap_name, p11h_name in direct_param_map.items():
- if ldap_name in ldap_key:
- unwrap_params[p11h_name] = ldap_key[ldap_name]
-
- # and some others needs conversion
-
- indirect_param_map = {
- "ipk11keytype": ("key_type", keytype_name2id),
- "ipawrappingmech": ("wrapping_mech", wrappingmech_name2id),
- }
-
- for ldap_name, rules in indirect_param_map.items():
- p11h_name, mapping = rules
- if ldap_name in ldap_key:
- unwrap_params[p11h_name] = mapping[ldap_key[ldap_name]]
-
- return unwrap_params
-
-
-class AbstractHSM(object):
- def _filter_replica_keys(self, all_keys):
- replica_keys = {}
- for key_id, key in all_keys.items():
- if not key['ipk11label'].startswith('dnssec-replica:'):
- continue
- replica_keys[key_id] = key
- return replica_keys
-
- def _filter_zone_keys(self, all_keys):
- zone_keys = {}
- for key_id, key in all_keys.items():
- if key['ipk11label'] == u'dnssec-master' \
- or key['ipk11label'].startswith('dnssec-replica:'):
- continue
- zone_keys[key_id] = key
- return zone_keys
diff --git a/ipapython/dnssec/bindmgr.py b/ipapython/dnssec/bindmgr.py
deleted file mode 100644
index 33d071f45..000000000
--- a/ipapython/dnssec/bindmgr.py
+++ /dev/null
@@ -1,217 +0,0 @@
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-from datetime import datetime
-import dns.name
-import errno
-import os
-import shutil
-import stat
-
-import ipalib.constants
-from ipapython.dn import DN
-from ipapython import ipa_log_manager, ipautil
-from ipaplatform.paths import paths
-
-from ipapython.dnssec.temp import TemporaryDirectory
-
-time_bindfmt = '%Y%m%d%H%M%S'
-
-# this daemon should run under ods:named user:group
-# user has to be ods because ODSMgr.py sends signal to ods-enforcerd
-FILE_PERM = (stat.S_IRUSR | stat.S_IRGRP | stat.S_IWGRP | stat.S_IWUSR)
-DIR_PERM = (stat.S_IRWXU | stat.S_IRWXG)
-
-class BINDMgr(object):
- """BIND key manager. It does LDAP->BIND key files synchronization.
-
- One LDAP object with idnsSecKey object class will produce
- single pair of BIND key files.
- """
- def __init__(self, api):
- self.api = api
- self.log = ipa_log_manager.log_mgr.get_logger(self)
- self.ldap_keys = {}
- self.modified_zones = set()
-
- def notify_zone(self, zone):
- cmd = ['rndc', 'sign', zone.to_text()]
- result = ipautil.run(cmd, capture_output=True)
- self.log.info('%s', result.output_log)
-
- def dn2zone_name(self, dn):
- """cn=KSK-20140813162153Z-cede9e182fc4af76c4bddbc19123a565,cn=keys,idnsname=test,cn=dns,dc=ipa,dc=example"""
- # verify that metadata object is under DNS sub-tree
- dn = DN(dn)
- container = DN(self.api.env.container_dns, self.api.env.basedn)
- idx = dn.rfind(container)
- assert idx != -1, 'Metadata object %s is not inside %s' % (dn, container)
- assert len(dn[idx - 1]) == 1, 'Multi-valued RDN as zone name is not supported'
- return dns.name.from_text(dn[idx - 1]['idnsname'])
-
- def time_ldap2bindfmt(self, str_val):
- dt = datetime.strptime(str_val, ipalib.constants.LDAP_GENERALIZED_TIME_FORMAT)
- return dt.strftime(time_bindfmt)
-
- def dates2params(self, ldap_attrs):
- """Convert LDAP timestamps to list of parameters suitable
- for dnssec-keyfromlabel utility"""
- attr2param = {'idnsseckeypublish': '-P',
- 'idnsseckeyactivate': '-A',
- 'idnsseckeyinactive': '-I',
- 'idnsseckeydelete': '-D'}
-
- params = []
- for attr, param in attr2param.items():
- params.append(param)
- if attr in ldap_attrs:
- assert len(ldap_attrs[attr]) == 1, 'Timestamp %s is expected to be single-valued' % attr
- params.append(self.time_ldap2bindfmt(ldap_attrs[attr][0]))
- else:
- params.append('none')
-
- return params
-
- def ldap_event(self, op, uuid, attrs):
- """Record single LDAP event - key addition, deletion or modification.
-
- Change is only recorded to memory.
- self.sync() has to be called to synchronize change to BIND."""
- assert op == 'add' or op == 'del' or op == 'mod'
- zone = self.dn2zone_name(attrs['dn'])
- self.modified_zones.add(zone)
- zone_keys = self.ldap_keys.setdefault(zone, {})
- if op == 'add':
- self.log.info('Key metadata %s added to zone %s' % (attrs['dn'], zone))
- zone_keys[uuid] = attrs
-
- elif op == 'del':
- self.log.info('Key metadata %s deleted from zone %s' % (attrs['dn'], zone))
- zone_keys.pop(uuid)
-
- elif op == 'mod':
- self.log.info('Key metadata %s updated in zone %s' % (attrs['dn'], zone))
- zone_keys[uuid] = attrs
-
- def install_key(self, zone, uuid, attrs, workdir):
- """Run dnssec-keyfromlabel on given LDAP object.
- :returns: base file name of output files, e.g. Kaaa.test.+008+19719"""
- self.log.info('attrs: %s', attrs)
- assert attrs.get('idnsseckeyzone', ['FALSE'])[0] == 'TRUE', \
- 'object %s is not a DNS zone key' % attrs['dn']
-
- uri = "%s;pin-source=%s" % (attrs['idnsSecKeyRef'][0], paths.DNSSEC_SOFTHSM_PIN)
- cmd = [paths.DNSSEC_KEYFROMLABEL, '-K', workdir, '-a', attrs['idnsSecAlgorithm'][0], '-l', uri]
- cmd += self.dates2params(attrs)
- if attrs.get('idnsSecKeySep', ['FALSE'])[0].upper() == 'TRUE':
- cmd += ['-f', 'KSK']
- if attrs.get('idnsSecKeyRevoke', ['FALSE'])[0].upper() == 'TRUE':
- cmd += ['-R', datetime.now().strftime(time_bindfmt)]
- cmd.append(zone.to_text())
-
- # keys has to be readable by ODS & named
- result = ipautil.run(cmd, capture_output=True)
- basename = result.output.strip()
- private_fn = "%s/%s.private" % (workdir, basename)
- os.chmod(private_fn, FILE_PERM)
- # this is useful mainly for debugging
- with open("%s/%s.uuid" % (workdir, basename), 'w') as uuid_file:
- uuid_file.write(uuid)
- with open("%s/%s.dn" % (workdir, basename), 'w') as dn_file:
- dn_file.write(attrs['dn'])
-
- def get_zone_dir_name(self, zone):
- """Escape zone name to form suitable for file-system.
-
- This method has to be equivalent to zr_get_zone_path()
- in bind-dyndb-ldap/zone_register.c."""
-
- if zone == dns.name.root:
- return "@"
-
- # strip final (empty) label
- zone = zone.relativize(dns.name.root)
- escaped = ""
- for label in zone:
- for char in label:
- c = ord(char)
- if ((c >= 0x30 and c <= 0x39) or # digit
- (c >= 0x41 and c <= 0x5A) or # uppercase
- (c >= 0x61 and c <= 0x7A) or # lowercase
- c == 0x2D or # hyphen
- c == 0x5F): # underscore
- if (c >= 0x41 and c <= 0x5A): # downcase
- c += 0x20
- escaped += chr(c)
- else:
- escaped += "%%%02X" % c
- escaped += '.'
-
- # strip trailing period
- return escaped[:-1]
-
- def sync_zone(self, zone):
- self.log.info('Synchronizing zone %s' % zone)
- zone_path = os.path.join(paths.BIND_LDAP_DNS_ZONE_WORKDIR,
- self.get_zone_dir_name(zone))
- try:
- os.makedirs(zone_path)
- except OSError as e:
- if e.errno != errno.EEXIST:
- raise e
-
- # fix HSM permissions
- # TODO: move out
- for prefix, dirs, files in os.walk(paths.DNSSEC_TOKENS_DIR, topdown=True):
- for name in dirs:
- fpath = os.path.join(prefix, name)
- self.log.debug('Fixing directory permissions: %s', fpath)
- os.chmod(fpath, DIR_PERM | stat.S_ISGID)
- for name in files:
- fpath = os.path.join(prefix, name)
- self.log.debug('Fixing file permissions: %s', fpath)
- os.chmod(fpath, FILE_PERM)
- # TODO: move out
-
- with TemporaryDirectory(zone_path) as tempdir:
- for uuid, attrs in self.ldap_keys[zone].items():
- self.install_key(zone, uuid, attrs, tempdir)
- # keys were generated in a temporary directory, swap directories
- target_dir = "%s/keys" % zone_path
- try:
- shutil.rmtree(target_dir)
- except OSError as e:
- if e.errno != errno.ENOENT:
- raise e
- shutil.move(tempdir, target_dir)
- os.chmod(target_dir, DIR_PERM)
-
- self.notify_zone(zone)
-
- def sync(self, dnssec_zones):
- """Synchronize list of zones in LDAP with BIND.
-
- dnssec_zones lists zones which should be processed. All other zones
- will be ignored even though they were modified using ldap_event().
-
- This filter is useful in cases where LDAP contains DNS zones which
- have old metadata objects and DNSSEC disabled. Such zones must be
- ignored to prevent errors while calling dnssec-keyfromlabel or rndc.
- """
- self.log.debug('Key metadata in LDAP: %s' % self.ldap_keys)
- self.log.debug('Zones modified but skipped during bindmgr.sync: %s',
- self.modified_zones - dnssec_zones)
- for zone in self.modified_zones.intersection(dnssec_zones):
- self.sync_zone(zone)
-
- self.modified_zones = set()
-
- def diff_zl(self, s1, s2):
- """Compute zones present in s1 but not present in s2.
-
- Returns: List of (uuid, name) tuples with zones present only in s1."""
- s1_extra = s1.uuids - s2.uuids
- removed = [(uuid, name) for (uuid, name) in s1.mapping.items()
- if uuid in s1_extra]
- return removed
diff --git a/ipapython/dnssec/keysyncer.py b/ipapython/dnssec/keysyncer.py
deleted file mode 100644
index 20039a068..000000000
--- a/ipapython/dnssec/keysyncer.py
+++ /dev/null
@@ -1,191 +0,0 @@
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-import ldap.dn
-import os
-
-import dns.name
-
-from ipaplatform.paths import paths
-from ipapython import ipautil
-
-from ipapython.dnssec.syncrepl import SyncReplConsumer
-from ipapython.dnssec.odsmgr import ODSMgr
-from ipapython.dnssec.bindmgr import BINDMgr
-
-SIGNING_ATTR = 'idnsSecInlineSigning'
-OBJCLASS_ATTR = 'objectClass'
-
-
-class KeySyncer(SyncReplConsumer):
- def __init__(self, *args, **kwargs):
- # hack
- self.api = kwargs['ipa_api']
- del kwargs['ipa_api']
-
- # DNSSEC master should have OpenDNSSEC installed
- # TODO: Is this the best way?
- if os.environ.get('ISMASTER', '0') == '1':
- self.ismaster = True
- self.odsmgr = ODSMgr()
- else:
- self.ismaster = False
-
- self.bindmgr = BINDMgr(self.api)
- self.init_done = False
- self.dnssec_zones = set()
- SyncReplConsumer.__init__(self, *args, **kwargs)
-
- def _get_objclass(self, attrs):
- """Get object class.
-
- Given set of attributes has to have exactly one supported object class.
- """
- supported_objclasses = set(['idnszone', 'idnsseckey', 'ipk11publickey'])
- present_objclasses = set([o.lower() for o in attrs[OBJCLASS_ATTR]]).intersection(supported_objclasses)
- assert len(present_objclasses) == 1, attrs[OBJCLASS_ATTR]
- return present_objclasses.pop()
-
- def __get_signing_attr(self, attrs):
- """Get SIGNING_ATTR from dictionary with LDAP zone attributes.
-
- Returned value is normalized to TRUE or FALSE, defaults to FALSE."""
- values = attrs.get(SIGNING_ATTR, ['FALSE'])
- assert len(values) == 1, '%s is expected to be single-valued' \
- % SIGNING_ATTR
- return values[0].upper()
-
- def __is_dnssec_enabled(self, attrs):
- """Test if LDAP DNS zone with given attributes is DNSSEC enabled."""
- return self.__get_signing_attr(attrs) == 'TRUE'
-
- def __is_replica_pubkey(self, attrs):
- vals = attrs.get('ipk11label', [])
- if len(vals) != 1:
- return False
- return vals[0].startswith('dnssec-replica:')
-
- def application_add(self, uuid, dn, newattrs):
- objclass = self._get_objclass(newattrs)
- if objclass == 'idnszone':
- self.zone_add(uuid, dn, newattrs)
- elif objclass == 'idnsseckey':
- self.key_meta_add(uuid, dn, newattrs)
- elif objclass == 'ipk11publickey' and \
- self.__is_replica_pubkey(newattrs):
- self.hsm_master_sync()
-
- def application_del(self, uuid, dn, oldattrs):
- objclass = self._get_objclass(oldattrs)
- if objclass == 'idnszone':
- self.zone_del(uuid, dn, oldattrs)
- elif objclass == 'idnsseckey':
- self.key_meta_del(uuid, dn, oldattrs)
- elif objclass == 'ipk11publickey' and \
- self.__is_replica_pubkey(oldattrs):
- self.hsm_master_sync()
-
- def application_sync(self, uuid, dn, newattrs, oldattrs):
- objclass = self._get_objclass(oldattrs)
- if objclass == 'idnszone':
- olddn = ldap.dn.str2dn(oldattrs['dn'])
- newdn = ldap.dn.str2dn(newattrs['dn'])
- assert olddn == newdn, 'modrdn operation is not supported'
-
- oldval = self.__get_signing_attr(oldattrs)
- newval = self.__get_signing_attr(newattrs)
- if oldval != newval:
- if self.__is_dnssec_enabled(newattrs):
- self.zone_add(uuid, olddn, newattrs)
- else:
- self.zone_del(uuid, olddn, oldattrs)
-
- elif objclass == 'idnsseckey':
- self.key_metadata_sync(uuid, dn, oldattrs, newattrs)
-
- elif objclass == 'ipk11publickey' and \
- self.__is_replica_pubkey(newattrs):
- self.hsm_master_sync()
-
- def syncrepl_refreshdone(self):
- self.log.info('Initial LDAP dump is done, sychronizing with ODS and BIND')
- self.init_done = True
- self.ods_sync()
- self.hsm_replica_sync()
- self.hsm_master_sync()
- self.bindmgr.sync(self.dnssec_zones)
-
- # idnsSecKey wrapper
- # Assumption: metadata points to the same key blob all the time,
- # i.e. it is not necessary to re-download blobs because of change in DNSSEC
- # metadata - DNSSEC flags or timestamps.
- def key_meta_add(self, uuid, dn, newattrs):
- self.hsm_replica_sync()
- self.bindmgr.ldap_event('add', uuid, newattrs)
- self.bindmgr_sync(self.dnssec_zones)
-
- def key_meta_del(self, uuid, dn, oldattrs):
- self.bindmgr.ldap_event('del', uuid, oldattrs)
- self.bindmgr_sync(self.dnssec_zones)
- self.hsm_replica_sync()
-
- def key_metadata_sync(self, uuid, dn, oldattrs, newattrs):
- self.bindmgr.ldap_event('mod', uuid, newattrs)
- self.bindmgr_sync(self.dnssec_zones)
-
- def bindmgr_sync(self, dnssec_zones):
- if self.init_done:
- self.bindmgr.sync(dnssec_zones)
-
- # idnsZone wrapper
- def zone_add(self, uuid, dn, newattrs):
- zone = dns.name.from_text(newattrs['idnsname'][0])
- if self.__is_dnssec_enabled(newattrs):
- self.dnssec_zones.add(zone)
- else:
- self.dnssec_zones.discard(zone)
-
- if not self.ismaster:
- return
-
- if self.__is_dnssec_enabled(newattrs):
- self.odsmgr.ldap_event('add', uuid, newattrs)
- self.ods_sync()
-
- def zone_del(self, uuid, dn, oldattrs):
- zone = dns.name.from_text(oldattrs['idnsname'][0])
- self.dnssec_zones.discard(zone)
-
- if not self.ismaster:
- return
-
- if self.__is_dnssec_enabled(oldattrs):
- self.odsmgr.ldap_event('del', uuid, oldattrs)
- self.ods_sync()
-
- def ods_sync(self):
- if not self.ismaster:
- return
-
- if self.init_done:
- self.odsmgr.sync()
-
- # triggered by modification to idnsSecKey objects
- def hsm_replica_sync(self):
- """Download keys from LDAP to local HSM."""
- if self.ismaster:
- return
- if not self.init_done:
- return
- ipautil.run([paths.IPA_DNSKEYSYNCD_REPLICA])
-
- # triggered by modification to ipk11PublicKey objects
- def hsm_master_sync(self):
- """Download replica keys from LDAP to local HSM
- & upload master and zone keys to LDAP."""
- if not self.ismaster:
- return
- if not self.init_done:
- return
- ipautil.run([paths.ODS_SIGNER, 'ipa-hsm-update'])
diff --git a/ipapython/dnssec/ldapkeydb.py b/ipapython/dnssec/ldapkeydb.py
deleted file mode 100644
index aa0413934..000000000
--- a/ipapython/dnssec/ldapkeydb.py
+++ /dev/null
@@ -1,450 +0,0 @@
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-from __future__ import print_function
-
-from binascii import hexlify
-import collections
-from pprint import pprint
-
-import ipalib
-from ipapython.dn import DN
-from ipapython import ipaldap
-from ipapython import ipa_log_manager
-
-from ipapython.dnssec.abshsm import (
- attrs_name2id,
- AbstractHSM,
- bool_attr_names,
- populate_pkcs11_metadata)
-from ipapython import p11helper as _ipap11helper
-import uuid
-
-def uri_escape(val):
- """convert val to %-notation suitable for ID component in URI"""
- assert len(val) > 0, "zero-length URI component detected"
- hexval = hexlify(val)
- out = '%'
- # pylint: disable=E1127
- out += '%'.join(hexval[i:i+2] for i in range(0, len(hexval), 2))
- return out
-
-def ldap_bool(val):
- if val == 'TRUE' or val is True:
- return True
- elif val == 'FALSE' or val is False:
- return False
- else:
- raise AssertionError('invalid LDAP boolean "%s"' % val)
-
-def get_default_attrs(object_classes):
- # object class -> default attribute values mapping
- defaults = {
- u'ipk11publickey': {
- 'ipk11copyable': True,
- 'ipk11derive': False,
- 'ipk11encrypt': False,
- 'ipk11local': True,
- 'ipk11modifiable': True,
- 'ipk11private': True,
- 'ipk11trusted': False,
- 'ipk11verify': True,
- 'ipk11verifyrecover': True,
- 'ipk11wrap': False
- },
- u'ipk11privatekey': {
- 'ipk11alwaysauthenticate': False,
- 'ipk11alwayssensitive': True,
- 'ipk11copyable': True,
- 'ipk11decrypt': False,
- 'ipk11derive': False,
- 'ipk11extractable': True,
- 'ipk11local': True,
- 'ipk11modifiable': True,
- 'ipk11neverextractable': False,
- 'ipk11private': True,
- 'ipk11sensitive': True,
- 'ipk11sign': True,
- 'ipk11signrecover': True,
- 'ipk11unwrap': False,
- 'ipk11wrapwithtrusted': False
- },
- u'ipk11secretkey': {
- 'ipk11alwaysauthenticate': False,
- 'ipk11alwayssensitive': True,
- 'ipk11copyable': True,
- 'ipk11decrypt': False,
- 'ipk11derive': False,
- 'ipk11encrypt': False,
- 'ipk11extractable': True,
- 'ipk11local': True,
- 'ipk11modifiable': True,
- 'ipk11neverextractable': False,
- 'ipk11private': True,
- 'ipk11sensitive': True,
- 'ipk11sign': False,
- 'ipk11trusted': False,
- 'ipk11unwrap': True,
- 'ipk11verify': False,
- 'ipk11wrap': True,
- 'ipk11wrapwithtrusted': False
- }
- }
-
- # get set of supported object classes
- present_clss = set()
- for cls in object_classes:
- present_clss.add(cls.lower())
- present_clss.intersection_update(set(defaults.keys()))
- if len(present_clss) <= 0:
- raise AssertionError('none of "%s" object classes are supported' %
- object_classes)
-
- result = {}
- for cls in present_clss:
- result.update(defaults[cls])
- return result
-
-
-class Key(collections.MutableMapping):
- """abstraction to hide LDAP entry weirdnesses:
- - non-normalized attribute names
- - boolean attributes returned as strings
- - planned entry deletion prevents subsequent use of the instance
- """
- def __init__(self, entry, ldap, ldapkeydb):
- self.entry = entry
- self._delentry = None # indicates that object was deleted
- self.ldap = ldap
- self.ldapkeydb = ldapkeydb
- self.log = ldap.log.getChild(__name__)
-
- def __assert_not_deleted(self):
- assert self.entry and not self._delentry, (
- "attempt to use to-be-deleted entry %s detected"
- % self._delentry.dn)
-
- def __getitem__(self, key):
- self.__assert_not_deleted()
- val = self.entry.single_value[key]
- if key.lower() in bool_attr_names:
- val = ldap_bool(val)
- return val
-
- def __setitem__(self, key, value):
- self.__assert_not_deleted()
- self.entry[key] = value
-
- def __delitem__(self, key):
- self.__assert_not_deleted()
- del self.entry[key]
-
- def __iter__(self):
- """generates list of ipa names of all PKCS#11 attributes present in the object"""
- self.__assert_not_deleted()
- for ipa_name in list(self.entry.keys()):
- lowercase = ipa_name.lower()
- if lowercase in attrs_name2id:
- yield lowercase
-
- def __len__(self):
- self.__assert_not_deleted()
- return len(self.entry)
-
- def __repr__(self):
- if self._delentry:
- return 'deleted entry: %s' % repr(self._delentry)
-
- sanitized = dict(self.entry)
- for attr in ['ipaPrivateKey', 'ipaPublicKey', 'ipk11publickeyinfo']:
- if attr in sanitized:
- del sanitized[attr]
- return repr(sanitized)
-
- def _cleanup_key(self):
- """remove default values from LDAP entry"""
- default_attrs = get_default_attrs(self.entry['objectclass'])
- empty = object()
- for attr in default_attrs:
- if self.get(attr, empty) == default_attrs[attr]:
- del self[attr]
-
- def _update_key(self):
- """remove default values from LDAP entry and write back changes"""
- if self._delentry:
- self._delete_key()
- return
-
- self._cleanup_key()
-
- try:
- self.ldap.update_entry(self.entry)
- except ipalib.errors.EmptyModlist:
- pass
-
- def _delete_key(self):
- """remove key metadata entry from LDAP
-
- After calling this, the python object is no longer valid and all
- subsequent method calls on it will fail.
- """
- assert not self.entry, (
- "Key._delete_key() called before Key.schedule_deletion()")
- assert self._delentry, "Key._delete_key() called more than once"
- self.log.debug('deleting key id 0x%s DN %s from LDAP',
- hexlify(self._delentry.single_value['ipk11id']),
- self._delentry.dn)
- self.ldap.delete_entry(self._delentry)
- self._delentry = None
- self.ldap = None
- self.ldapkeydb = None
-
- def schedule_deletion(self):
- """schedule key deletion from LDAP
-
- Calling schedule_deletion() will make this object incompatible with
- normal Key. After that the object must not be read or modified.
- Key metadata will be actually deleted when LdapKeyDB.flush() is called.
- """
- assert not self._delentry, (
- "Key.schedule_deletion() called more than once")
- self._delentry = self.entry
- self.entry = None
-
-
-class ReplicaKey(Key):
- # TODO: object class assert
- def __init__(self, entry, ldap, ldapkeydb):
- super(ReplicaKey, self).__init__(entry, ldap, ldapkeydb)
-
-class MasterKey(Key):
- # TODO: object class assert
- def __init__(self, entry, ldap, ldapkeydb):
- super(MasterKey, self).__init__(entry, ldap, ldapkeydb)
-
- @property
- def wrapped_entries(self):
- """LDAP entires with wrapped data
-
- One entry = one blob + ipaWrappingKey pointer to unwrapping key"""
-
- keys = []
- if 'ipaSecretKeyRef' not in self.entry:
- return keys
-
- for dn in self.entry['ipaSecretKeyRef']:
- try:
- obj = self.ldap.get_entry(dn)
- keys.append(obj)
- except ipalib.errors.NotFound:
- continue
-
- return keys
-
- def add_wrapped_data(self, data, wrapping_mech, replica_key_id):
- wrapping_key_uri = 'pkcs11:id=%s;type=public' \
- % uri_escape(replica_key_id)
- # TODO: replace this with 'autogenerate' to prevent collisions
- uuid_rdn = DN('ipk11UniqueId=%s' % uuid.uuid1())
- entry_dn = DN(uuid_rdn, self.ldapkeydb.base_dn)
- entry = self.ldap.make_entry(entry_dn,
- objectClass=['ipaSecretKeyObject', 'ipk11Object'],
- ipaSecretKey=data,
- ipaWrappingKey=wrapping_key_uri,
- ipaWrappingMech=wrapping_mech)
-
- self.log.info('adding master key 0x%s wrapped with replica key 0x%s to %s',
- hexlify(self['ipk11id']),
- hexlify(replica_key_id),
- entry_dn)
- self.ldap.add_entry(entry)
- if 'ipaSecretKeyRef' not in self.entry:
- self.entry['objectClass'] += ['ipaSecretKeyRefObject']
- self.entry.setdefault('ipaSecretKeyRef', []).append(entry_dn)
-
-
-class LdapKeyDB(AbstractHSM):
- def __init__(self, log, ldap, base_dn):
- self.ldap = ldap
- self.base_dn = base_dn
- self.log = log
- self.cache_replica_pubkeys_wrap = None
- self.cache_masterkeys = None
- self.cache_zone_keypairs = None
-
- def _get_key_dict(self, key_type, ldap_filter):
- try:
- objs = self.ldap.get_entries(base_dn=self.base_dn,
- filter=ldap_filter)
- except ipalib.errors.NotFound:
- return {}
-
- keys = {}
- for o in objs:
- # add default values not present in LDAP
- key = key_type(o, self.ldap, self)
- default_attrs = get_default_attrs(key.entry['objectclass'])
- for attr in default_attrs:
- key.setdefault(attr, default_attrs[attr])
-
- assert 'ipk11id' in key, 'key is missing ipk11Id in %s' % key.entry.dn
- key_id = key['ipk11id']
- assert key_id not in keys, 'duplicate ipk11Id=0x%s in "%s" and "%s"' % (hexlify(key_id), key.entry.dn, keys[key_id].entry.dn)
- assert 'ipk11label' in key, 'key "%s" is missing ipk11Label' % key.entry.dn
- assert 'objectclass' in key.entry, 'key "%s" is missing objectClass attribute' % key.entry.dn
-
- keys[key_id] = key
-
- self._update_keys()
- return keys
-
- def _update_keys(self):
- for cache in [self.cache_masterkeys, self.cache_replica_pubkeys_wrap,
- self.cache_zone_keypairs]:
- if cache:
- for key in cache.values():
- key._update_key()
-
- def flush(self):
- """write back content of caches to LDAP"""
- self._update_keys()
- self.cache_masterkeys = None
- self.cache_replica_pubkeys_wrap = None
- self.cache_zone_keypairs = None
-
- def _import_keys_metadata(self, source_keys):
- """import key metadata from Key-compatible objects
-
- metadata from multiple source keys can be imported into single LDAP
- object
-
- :param: source_keys is iterable of (Key object, PKCS#11 object class)"""
-
- entry_dn = DN('ipk11UniqueId=autogenerate', self.base_dn)
- entry = self.ldap.make_entry(entry_dn, objectClass=['ipk11Object'])
- new_key = Key(entry, self.ldap, self)
-
- for source_key, pkcs11_class in source_keys:
- if pkcs11_class == _ipap11helper.KEY_CLASS_SECRET_KEY:
- entry['objectClass'].append('ipk11SecretKey')
- elif pkcs11_class == _ipap11helper.KEY_CLASS_PUBLIC_KEY:
- entry['objectClass'].append('ipk11PublicKey')
- elif pkcs11_class == _ipap11helper.KEY_CLASS_PRIVATE_KEY:
- entry['objectClass'].append('ipk11PrivateKey')
- else:
- raise AssertionError('unsupported object class %s' % pkcs11_class)
-
- populate_pkcs11_metadata(source_key, new_key)
- new_key._cleanup_key()
- return new_key
-
- def import_master_key(self, mkey):
- new_key = self._import_keys_metadata(
- [(mkey, _ipap11helper.KEY_CLASS_SECRET_KEY)])
- self.ldap.add_entry(new_key.entry)
- self.log.debug('imported master key metadata: %s', new_key.entry)
-
- def import_zone_key(self, pubkey, pubkey_data, privkey,
- privkey_wrapped_data, wrapping_mech, master_key_id):
- new_key = self._import_keys_metadata(
- [(pubkey, _ipap11helper.KEY_CLASS_PUBLIC_KEY),
- (privkey, _ipap11helper.KEY_CLASS_PRIVATE_KEY)])
-
- new_key.entry['objectClass'].append('ipaPrivateKeyObject')
- new_key.entry['ipaPrivateKey'] = privkey_wrapped_data
- new_key.entry['ipaWrappingKey'] = 'pkcs11:id=%s;type=secret-key' \
- % uri_escape(master_key_id)
- new_key.entry['ipaWrappingMech'] = wrapping_mech
-
- new_key.entry['objectClass'].append('ipaPublicKeyObject')
- new_key.entry['ipaPublicKey'] = pubkey_data
-
- self.ldap.add_entry(new_key.entry)
- self.log.debug('imported zone key id: 0x%s', hexlify(new_key['ipk11id']))
-
- @property
- def replica_pubkeys_wrap(self):
- if self.cache_replica_pubkeys_wrap:
- return self.cache_replica_pubkeys_wrap
-
- keys = self._filter_replica_keys(
- self._get_key_dict(ReplicaKey,
- '(&(objectClass=ipk11PublicKey)(ipk11Wrap=TRUE)(objectClass=ipaPublicKeyObject))'))
-
- self.cache_replica_pubkeys_wrap = keys
- return keys
-
- @property
- def master_keys(self):
- if self.cache_masterkeys:
- return self.cache_masterkeys
-
- keys = self._get_key_dict(MasterKey,
- '(&(objectClass=ipk11SecretKey)(|(ipk11UnWrap=TRUE)(!(ipk11UnWrap=*)))(ipk11Label=dnssec-master))')
- for key in keys.values():
- prefix = 'dnssec-master'
- assert key['ipk11label'] == prefix, \
- 'secret key dn="%s" ipk11id=0x%s ipk11label="%s" with ipk11UnWrap = TRUE does not have '\
- '"%s" key label' % (
- key.entry.dn,
- hexlify(key['ipk11id']),
- str(key['ipk11label']),
- prefix)
-
- self.cache_masterkeys = keys
- return keys
-
- @property
- def zone_keypairs(self):
- if self.cache_zone_keypairs:
- return self.cache_zone_keypairs
-
- self.cache_zone_keypairs = self._filter_zone_keys(
- self._get_key_dict(Key,
- '(&(objectClass=ipk11PrivateKey)(objectClass=ipaPrivateKeyObject)(objectClass=ipk11PublicKey)(objectClass=ipaPublicKeyObject))'))
-
- return self.cache_zone_keypairs
-
-if __name__ == '__main__':
- # this is debugging mode
- # print information we think are useful to stdout
- # other garbage goes via logger to stderr
- ipa_log_manager.standard_logging_setup(debug=True)
- log = ipa_log_manager.root_logger
-
- # IPA framework initialization
- ipalib.api.bootstrap(in_server=True, log=None) # no logging to file
- ipalib.api.finalize()
-
- # LDAP initialization
- dns_dn = DN(ipalib.api.env.container_dns, ipalib.api.env.basedn)
- ldap = ipaldap.LDAPClient(ipalib.api.env.ldap_uri)
- log.debug('Connecting to LDAP')
- # GSSAPI will be used, used has to be kinited already
- ldap.gssapi_bind()
- log.debug('Connected')
-
- ldapkeydb = LdapKeyDB(log, ldap, DN(('cn', 'keys'), ('cn', 'sec'),
- ipalib.api.env.container_dns,
- ipalib.api.env.basedn))
-
- print('replica public keys: CKA_WRAP = TRUE')
- print('====================================')
- for pubkey_id, pubkey in ldapkeydb.replica_pubkeys_wrap.items():
- print(hexlify(pubkey_id))
- pprint(pubkey)
-
- print('')
- print('master keys')
- print('===========')
- for mkey_id, mkey in ldapkeydb.master_keys.items():
- print(hexlify(mkey_id))
- pprint(mkey)
-
- print('')
- print('zone key pairs')
- print('==============')
- for key_id, key in ldapkeydb.zone_keypairs.items():
- print(hexlify(key_id))
- pprint(key)
diff --git a/ipapython/dnssec/localhsm.py b/ipapython/dnssec/localhsm.py
deleted file mode 100755
index befe08aec..000000000
--- a/ipapython/dnssec/localhsm.py
+++ /dev/null
@@ -1,227 +0,0 @@
-#!/usr/bin/python2
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-from __future__ import print_function
-
-from binascii import hexlify
-import collections
-import logging
-import os
-from pprint import pprint
-
-from ipaplatform.paths import paths
-
-from ipapython import p11helper as _ipap11helper
-from ipapython.dnssec.abshsm import (attrs_name2id, attrs_id2name, AbstractHSM,
- keytype_id2name, keytype_name2id,
- ldap2p11helper_api_params)
-
-private_key_api_params = set(["label", "id", "data", "unwrapping_key",
- "wrapping_mech", "key_type", "cka_always_authenticate", "cka_copyable",
- "cka_decrypt", "cka_derive", "cka_extractable", "cka_modifiable",
- "cka_private", "cka_sensitive", "cka_sign", "cka_sign_recover",
- "cka_unwrap", "cka_wrap_with_trusted"])
-
-public_key_api_params = set(["label", "id", "data", "cka_copyable",
- "cka_derive", "cka_encrypt", "cka_modifiable", "cka_private",
- "cka_trusted", "cka_verify", "cka_verify_recover", "cka_wrap"])
-
-class Key(collections.MutableMapping):
- def __init__(self, p11, handle):
- self.p11 = p11
- self.handle = handle
- # sanity check CKA_ID and CKA_LABEL
- try:
- cka_id = self.p11.get_attribute(handle, _ipap11helper.CKA_ID)
- assert len(cka_id) != 0, 'ipk11id length should not be 0'
- except _ipap11helper.NotFound:
- raise _ipap11helper.NotFound('key without ipk11id: handle %s' % handle)
-
- try:
- cka_label = self.p11.get_attribute(handle, _ipap11helper.CKA_LABEL)
- assert len(cka_label) != 0, 'ipk11label length should not be 0'
-
- except _ipap11helper.NotFound:
- raise _ipap11helper.NotFound('key without ipk11label: id 0x%s'
- % hexlify(cka_id))
-
- def __getitem__(self, key):
- key = key.lower()
- try:
- value = self.p11.get_attribute(self.handle, attrs_name2id[key])
- if key == 'ipk11keytype':
- value = keytype_id2name[value]
- return value
- except _ipap11helper.NotFound:
- raise KeyError()
-
- def __setitem__(self, key, value):
- key = key.lower()
- if key == 'ipk11keytype':
- value = keytype_name2id[value]
-
- return self.p11.set_attribute(self.handle, attrs_name2id[key], value)
-
- def __delitem__(self, key):
- raise _ipap11helper.P11HelperException('__delitem__ is not supported')
-
- def __iter__(self):
- """generates list of ipa names of all attributes present in the object"""
- for pkcs11_id, ipa_name in attrs_id2name.items():
- try:
- self.p11.get_attribute(self.handle, pkcs11_id)
- except _ipap11helper.NotFound:
- continue
-
- yield ipa_name
-
- def __len__(self):
- cnt = 0
- for _attr in self:
- cnt += 1
- return cnt
-
- def __str__(self):
- return str(dict(self))
-
- def __repr__(self):
- return self.__str__()
-
-class LocalHSM(AbstractHSM):
- def __init__(self, library, slot, pin):
- self.cache_replica_pubkeys = None
- self.p11 = _ipap11helper.P11_Helper(slot, pin, library)
- self.log = logging.getLogger()
-
- def __del__(self):
- self.p11.finalize()
-
- def find_keys(self, **kwargs):
- """Return dict with Key objects matching given criteria.
-
- CKA_ID is used as key so all matching objects have to have unique ID."""
-
- # this is a hack for old p11-kit URI parser
- # see https://bugs.freedesktop.org/show_bug.cgi?id=85057
- if 'uri' in kwargs:
- kwargs['uri'] = kwargs['uri'].replace('type=', 'object-type=')
-
- handles = self.p11.find_keys(**kwargs)
- keys = {}
- for h in handles:
- key = Key(self.p11, h)
- o_id = key['ipk11id']
- assert o_id not in keys, 'duplicate ipk11Id = 0x%s; keys = %s' % (
- hexlify(o_id), keys)
- keys[o_id] = key
-
- return keys
-
- @property
- def replica_pubkeys(self):
- return self._filter_replica_keys(
- self.find_keys(objclass=_ipap11helper.KEY_CLASS_PUBLIC_KEY))
-
- @property
- def replica_pubkeys_wrap(self):
- return self._filter_replica_keys(
- self.find_keys(objclass=_ipap11helper.KEY_CLASS_PUBLIC_KEY,
- cka_wrap=True))
-
- @property
- def master_keys(self):
- """Get all usable DNSSEC master keys"""
- keys = self.find_keys(objclass=_ipap11helper.KEY_CLASS_SECRET_KEY, label=u'dnssec-master', cka_unwrap=True)
-
- for key in keys.values():
- prefix = 'dnssec-master'
- assert key['ipk11label'] == prefix, \
- 'secret key ipk11id=0x%s ipk11label="%s" with ipk11UnWrap = TRUE does not have '\
- '"%s" key label' % (hexlify(key['ipk11id']),
- str(key['ipk11label']), prefix)
-
- return keys
-
- @property
- def active_master_key(self):
- """Get one active DNSSEC master key suitable for key wrapping"""
- keys = self.find_keys(objclass=_ipap11helper.KEY_CLASS_SECRET_KEY,
- label=u'dnssec-master', cka_wrap=True, cka_unwrap=True)
- assert len(keys) > 0, "DNSSEC master key with UN/WRAP = TRUE not found"
- return keys.popitem()[1]
-
- @property
- def zone_pubkeys(self):
- return self._filter_zone_keys(
- self.find_keys(objclass=_ipap11helper.KEY_CLASS_PUBLIC_KEY))
-
- @property
- def zone_privkeys(self):
- return self._filter_zone_keys(
- self.find_keys(objclass=_ipap11helper.KEY_CLASS_PRIVATE_KEY))
-
-
- def import_public_key(self, source, data):
- params = ldap2p11helper_api_params(source)
- # filter out params inappropriate for public keys
- for par in set(params).difference(public_key_api_params):
- del params[par]
- params['data'] = data
-
- h = self.p11.import_public_key(**params)
- return Key(self.p11, h)
-
- def import_private_key(self, source, data, unwrapping_key):
- params = ldap2p11helper_api_params(source)
- # filter out params inappropriate for private keys
- for par in set(params).difference(private_key_api_params):
- del params[par]
- params['data'] = data
- params['unwrapping_key'] = unwrapping_key.handle
-
- h = self.p11.import_wrapped_private_key(**params)
- return Key(self.p11, h)
-
-
-
-if __name__ == '__main__':
- if 'SOFTHSM2_CONF' not in os.environ:
- os.environ['SOFTHSM2_CONF'] = paths.DNSSEC_SOFTHSM2_CONF
- localhsm = LocalHSM(paths.LIBSOFTHSM2_SO, 0,
- open(paths.DNSSEC_SOFTHSM_PIN).read())
-
- print('replica public keys: CKA_WRAP = TRUE')
- print('====================================')
- for pubkey_id, pubkey in localhsm.replica_pubkeys_wrap.items():
- print(hexlify(pubkey_id))
- pprint(pubkey)
-
- print('')
- print('replica public keys: all')
- print('========================')
- for pubkey_id, pubkey in localhsm.replica_pubkeys.items():
- print(hexlify(pubkey_id))
- pprint(pubkey)
-
- print('')
- print('master keys')
- print('===========')
- for mkey_id, mkey in localhsm.master_keys.items():
- print(hexlify(mkey_id))
- pprint(mkey)
-
- print('')
- print('zone public keys')
- print('================')
- for key_id, key in localhsm.zone_pubkeys.items():
- print(hexlify(key_id))
- pprint(key)
-
- print('')
- print('zone private keys')
- print('=================')
- for key_id, key in localhsm.zone_privkeys.items():
- print(hexlify(key_id))
- pprint(key)
diff --git a/ipapython/dnssec/odsmgr.py b/ipapython/dnssec/odsmgr.py
deleted file mode 100644
index 0308408e0..000000000
--- a/ipapython/dnssec/odsmgr.py
+++ /dev/null
@@ -1,210 +0,0 @@
-#!/usr/bin/python2
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-import dns.name
-try:
- from xml.etree import cElementTree as etree
-except ImportError:
- from xml.etree import ElementTree as etree
-
-from ipapython import ipa_log_manager, ipautil
-
-# hack: zone object UUID is stored as path to imaginary zone file
-ENTRYUUID_PREFIX = "/var/lib/ipa/dns/zone/entryUUID/"
-ENTRYUUID_PREFIX_LEN = len(ENTRYUUID_PREFIX)
-
-
-class ZoneListReader(object):
- def __init__(self):
- self.names = set() # dns.name
- self.uuids = set() # UUID strings
- self.mapping = dict() # {UUID: dns.name}
- self.log = ipa_log_manager.log_mgr.get_logger(self)
-
- def _add_zone(self, name, zid):
- """Add zone & UUID to internal structures.
-
- Zone with given name and UUID must not exist."""
- # detect duplicate zone names
- name = dns.name.from_text(name)
- assert name not in self.names, \
- 'duplicate name (%s, %s) vs. %s' % (name, zid, self.mapping)
- # duplicate non-None zid is not allowed
- assert not zid or zid not in self.uuids, \
- 'duplicate UUID (%s, %s) vs. %s' % (name, zid, self.mapping)
-
- self.names.add(name)
- self.uuids.add(zid)
- self.mapping[zid] = name
-
- def _del_zone(self, name, zid):
- """Remove zone & UUID from internal structures.
-
- Zone with given name and UUID must exist.
- """
- name = dns.name.from_text(name)
- assert zid is not None
- assert name in self.names, \
- 'name (%s, %s) does not exist in %s' % (name, zid, self.mapping)
- assert zid in self.uuids, \
- 'UUID (%s, %s) does not exist in %s' % (name, zid, self.mapping)
- assert zid in self.mapping and name == self.mapping[zid], \
- 'pair {%s: %s} does not exist in %s' % (zid, name, self.mapping)
-
- self.names.remove(name)
- self.uuids.remove(zid)
- del self.mapping[zid]
-
-
-class ODSZoneListReader(ZoneListReader):
- """One-shot parser for ODS zonelist.xml."""
- def __init__(self, zonelist_text):
- super(ODSZoneListReader, self).__init__()
- root = etree.fromstring(zonelist_text)
- self._parse_zonelist(root)
-
- def _parse_zonelist(self, root):
- """iterate over Zone elements with attribute 'name' and
- add IPA zones to self.zones"""
- if not root.tag == 'ZoneList':
- raise ValueError(root.tag)
- for zone_xml in root.findall('./Zone[@name]'):
- name, zid = self._parse_ipa_zone(zone_xml)
- self._add_zone(name, zid)
-
- def _parse_ipa_zone(self, zone_xml):
- """Extract zone name, input adapter and detect IPA zones.
-
- IPA zones have contains Adapters/Input/Adapter element with
- attribute type = "File" and with value prefixed with ENTRYUUID_PREFIX.
-
- Returns:
- tuple (zone name, ID)
- """
- name = zone_xml.get('name')
- zids = []
- for in_adapter in zone_xml.findall(
- './Adapters/Input/Adapter[@type="File"]'):
- path = in_adapter.text
- if path.startswith(ENTRYUUID_PREFIX):
- # strip prefix from path
- zids.append(path[ENTRYUUID_PREFIX_LEN:])
-
- if len(zids) != 1:
- raise ValueError('only IPA zones are supported: {}'.format(
- etree.tostring(zone_xml)))
-
- return name, zids[0]
-
-
-class LDAPZoneListReader(ZoneListReader):
- def __init__(self):
- super(LDAPZoneListReader, self).__init__()
-
- def process_ipa_zone(self, op, uuid, zone_ldap):
- assert (op == 'add' or op == 'del'), 'unsupported op %s' % op
- assert uuid is not None
- assert 'idnsname' in zone_ldap, \
- 'LDAP zone UUID %s without idnsName' % uuid
- assert len(zone_ldap['idnsname']) == 1, \
- 'LDAP zone UUID %s with len(idnsname) != 1' % uuid
-
- if op == 'add':
- self._add_zone(zone_ldap['idnsname'][0], uuid)
- elif op == 'del':
- self._del_zone(zone_ldap['idnsname'][0], uuid)
-
-
-class ODSMgr(object):
- """OpenDNSSEC zone manager. It does LDAP->ODS synchronization.
-
- Zones with idnsSecInlineSigning attribute = TRUE in LDAP are added
- or deleted from ODS as necessary. ODS->LDAP key synchronization
- has to be solved seperatelly.
- """
- def __init__(self):
- self.log = ipa_log_manager.log_mgr.get_logger(self)
- self.zl_ldap = LDAPZoneListReader()
-
- def ksmutil(self, params):
- """Call ods-ksmutil with given parameters and return stdout.
-
- Raises CalledProcessError if returncode != 0.
- """
- cmd = ['ods-ksmutil'] + params
- result = ipautil.run(cmd, capture_output=True)
- return result.output
-
- def get_ods_zonelist(self):
- stdout = self.ksmutil(['zonelist', 'export'])
- reader = ODSZoneListReader(stdout)
- return reader
-
- def add_ods_zone(self, uuid, name):
- zone_path = '%s%s' % (ENTRYUUID_PREFIX, uuid)
- cmd = ['zone', 'add', '--zone', str(name), '--input', zone_path]
- output = self.ksmutil(cmd)
- self.log.info(output)
- self.notify_enforcer()
-
- def del_ods_zone(self, name):
- # ods-ksmutil blows up if zone name has period at the end
- name = name.relativize(dns.name.root)
- # detect if name is root zone
- if name == dns.name.empty:
- name = dns.name.root
- cmd = ['zone', 'delete', '--zone', str(name)]
- output = self.ksmutil(cmd)
- self.log.info(output)
- self.notify_enforcer()
- self.cleanup_signer(name)
-
- def notify_enforcer(self):
- cmd = ['notify']
- output = self.ksmutil(cmd)
- self.log.info(output)
-
- def cleanup_signer(self, zone_name):
- cmd = ['ods-signer', 'ldap-cleanup', str(zone_name)]
- output = ipautil.run(cmd, capture_output=True)
- self.log.info(output)
-
- def ldap_event(self, op, uuid, attrs):
- """Record single LDAP event - zone addition or deletion.
-
- Change is only recorded to memory.
- self.sync() have to be called to synchronize change to ODS."""
- assert op == 'add' or op == 'del'
- self.zl_ldap.process_ipa_zone(op, uuid, attrs)
- self.log.debug("LDAP zones: %s", self.zl_ldap.mapping)
-
- def sync(self):
- """Synchronize list of zones in LDAP with ODS."""
- zl_ods = self.get_ods_zonelist()
- self.log.debug("ODS zones: %s", zl_ods.mapping)
- removed = self.diff_zl(zl_ods, self.zl_ldap)
- self.log.info("Zones removed from LDAP: %s", removed)
- added = self.diff_zl(self.zl_ldap, zl_ods)
- self.log.info("Zones added to LDAP: %s", added)
- for (uuid, name) in removed:
- self.del_ods_zone(name)
- for (uuid, name) in added:
- self.add_ods_zone(uuid, name)
-
- def diff_zl(self, s1, s2):
- """Compute zones present in s1 but not present in s2.
-
- Returns: List of (uuid, name) tuples with zones present only in s1."""
- s1_extra = s1.uuids - s2.uuids
- removed = [(uuid, name) for (uuid, name) in s1.mapping.items()
- if uuid in s1_extra]
- return removed
-
-
-if __name__ == '__main__':
- ipa_log_manager.standard_logging_setup(debug=True)
- ods = ODSMgr()
- reader = ods.get_ods_zonelist()
- ipa_log_manager.root_logger.info('ODS zones: %s', reader.mapping)
diff --git a/ipapython/dnssec/syncrepl.py b/ipapython/dnssec/syncrepl.py
deleted file mode 100644
index e197670ad..000000000
--- a/ipapython/dnssec/syncrepl.py
+++ /dev/null
@@ -1,115 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-"""
-This script implements a syncrepl consumer which syncs data from server
-to a local dict.
-"""
-
-# Import the python-ldap modules
-import ldap
-# Import specific classes from python-ldap
-from ldap.cidict import cidict
-from ldap.ldapobject import ReconnectLDAPObject
-from ldap.syncrepl import SyncreplConsumer
-
-from ipapython import ipa_log_manager
-
-
-class SyncReplConsumer(ReconnectLDAPObject, SyncreplConsumer):
- """
- Syncrepl Consumer interface
- """
-
- def __init__(self, *args, **kwargs):
- self.log = ipa_log_manager.log_mgr.get_logger(self)
- # Initialise the LDAP Connection first
- ldap.ldapobject.ReconnectLDAPObject.__init__(self, *args, **kwargs)
- # Now prepare the data store
- self.__data = cidict()
- self.__data['uuids'] = cidict()
- # We need this for later internal use
- self.__presentUUIDs = cidict()
-
- def close_db(self):
- # This is useless for dict
- pass
-
- def syncrepl_get_cookie(self):
- if 'cookie' in self.__data:
- cookie = self.__data['cookie']
- self.log.debug('Current cookie is: %s', cookie)
- return cookie
- else:
- self.log.debug('Current cookie is: None (not received yet)')
-
- def syncrepl_set_cookie(self, cookie):
- self.log.debug('New cookie is: %s', cookie)
- self.__data['cookie'] = cookie
-
- def syncrepl_entry(self, dn, attributes, uuid):
- attributes = cidict(attributes)
- # First we determine the type of change we have here
- # (and store away the previous data for later if needed)
- previous_attributes = cidict()
- if uuid in self.__data['uuids']:
- change_type = 'modify'
- previous_attributes = self.__data['uuids'][uuid]
- else:
- change_type = 'add'
- # Now we store our knowledge of the existence of this entry
- # (including the DN as an attribute for convenience)
- attributes['dn'] = dn
- self.__data['uuids'][uuid] = attributes
- # Debugging
- self.log.debug('Detected %s of entry: %s %s', change_type, dn, uuid)
- if change_type == 'modify':
- self.application_sync(uuid, dn, attributes, previous_attributes)
- else:
- self.application_add(uuid, dn, attributes)
-
- def syncrepl_delete(self, uuids):
- # Make sure we know about the UUID being deleted, just in case...
- uuids = [uuid for uuid in uuids if uuid in self.__data['uuids']]
- # Delete all the UUID values we know of
- for uuid in uuids:
- attributes = self.__data['uuids'][uuid]
- dn = attributes['dn']
- self.log.debug('Detected deletion of entry: %s %s', dn, uuid)
- self.application_del(uuid, dn, attributes)
- del self.__data['uuids'][uuid]
-
- def syncrepl_present(self, uuids, refreshDeletes=False):
- # If we have not been given any UUID values,
- # then we have received all the present controls...
- if uuids is None:
- # We only do things if refreshDeletes is false
- # as the syncrepl extension will call syncrepl_delete instead
- # when it detects a delete notice
- if refreshDeletes is False:
- deletedEntries = [uuid for uuid in self.__data['uuids'].keys()
- if uuid not in self.__presentUUIDs]
- self.syncrepl_delete(deletedEntries)
- # Phase is now completed, reset the list
- self.__presentUUIDs = {}
- else:
- # Note down all the UUIDs we have been sent
- for uuid in uuids:
- self.__presentUUIDs[uuid] = True
-
- def application_add(self, uuid, dn, attributes):
- self.log.info('Performing application add for: %s %s', dn, uuid)
- self.log.debug('New attributes: %s', attributes)
- return True
-
- def application_sync(self, uuid, dn, attributes, previous_attributes):
- self.log.info('Performing application sync for: %s %s', dn, uuid)
- self.log.debug('Old attributes: %s', previous_attributes)
- self.log.debug('New attributes: %s', attributes)
- return True
-
- def application_del(self, uuid, dn, previous_attributes):
- self.log.info('Performing application delete for: %s %s', dn, uuid)
- self.log.debug('Old attributes: %s', previous_attributes)
- return True
diff --git a/ipapython/dnssec/temp.py b/ipapython/dnssec/temp.py
deleted file mode 100644
index e97d3a0b8..000000000
--- a/ipapython/dnssec/temp.py
+++ /dev/null
@@ -1,22 +0,0 @@
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-import errno
-import shutil
-import tempfile
-
-class TemporaryDirectory(object):
- def __init__(self, root):
- self.root = root
-
- def __enter__(self):
- self.name = tempfile.mkdtemp(dir=self.root)
- return self.name
-
- def __exit__(self, exc_type, exc_value, traceback):
- try:
- shutil.rmtree(self.name)
- except OSError as e:
- if e.errno != errno.ENOENT:
- raise
diff --git a/ipapython/p11helper.py b/ipapython/p11helper.py
deleted file mode 100644
index 5963c6d71..000000000
--- a/ipapython/p11helper.py
+++ /dev/null
@@ -1,1772 +0,0 @@
-#
-# Copyright (C) 2014 FreeIPA Contributors see COPYING for license
-#
-
-import random
-import ctypes.util
-import binascii
-
-import six
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric import dsa, ec, rsa
-from cffi import FFI
-
-if six.PY3:
- unicode = str
-
-
-_ffi = FFI()
-
-_ffi.cdef('''
-/* p11-kit/pkcs11.h */
-
-typedef unsigned long CK_FLAGS;
-
-struct _CK_VERSION
-{
- unsigned char major;
- unsigned char minor;
-};
-
-typedef unsigned long CK_SLOT_ID;
-
-typedef unsigned long CK_SESSION_HANDLE;
-
-typedef unsigned long CK_USER_TYPE;
-
-typedef unsigned long CK_OBJECT_HANDLE;
-
-typedef unsigned long CK_OBJECT_CLASS;
-
-typedef unsigned long CK_KEY_TYPE;
-
-typedef unsigned long CK_ATTRIBUTE_TYPE;
-
-struct _CK_ATTRIBUTE
-{
- CK_ATTRIBUTE_TYPE type;
- void *pValue;
- unsigned long ulValueLen;
-};
-
-typedef unsigned long CK_MECHANISM_TYPE;
-
-struct _CK_MECHANISM
-{
- CK_MECHANISM_TYPE mechanism;
- void *pParameter;
- unsigned long ulParameterLen;
-};
-
-typedef unsigned long CK_RV;
-
-typedef ... *CK_NOTIFY;
-
-struct _CK_FUNCTION_LIST;
-
-typedef CK_RV (*CK_C_Initialize) (void *init_args);
-typedef CK_RV (*CK_C_Finalize) (void *pReserved);
-typedef ... *CK_C_GetInfo;
-typedef ... *CK_C_GetFunctionList;
-CK_RV C_GetFunctionList (struct _CK_FUNCTION_LIST **function_list);
-typedef ... *CK_C_GetSlotList;
-typedef ... *CK_C_GetSlotInfo;
-typedef ... *CK_C_GetTokenInfo;
-typedef ... *CK_C_WaitForSlotEvent;
-typedef ... *CK_C_GetMechanismList;
-typedef ... *CK_C_GetMechanismInfo;
-typedef ... *CK_C_InitToken;
-typedef ... *CK_C_InitPIN;
-typedef ... *CK_C_SetPIN;
-typedef CK_RV (*CK_C_OpenSession) (CK_SLOT_ID slotID, CK_FLAGS flags,
- void *application, CK_NOTIFY notify,
- CK_SESSION_HANDLE *session);
-typedef CK_RV (*CK_C_CloseSession) (CK_SESSION_HANDLE session);
-typedef ... *CK_C_CloseAllSessions;
-typedef ... *CK_C_GetSessionInfo;
-typedef ... *CK_C_GetOperationState;
-typedef ... *CK_C_SetOperationState;
-typedef CK_RV (*CK_C_Login) (CK_SESSION_HANDLE session, CK_USER_TYPE user_type,
- unsigned char *pin, unsigned long pin_len);
-typedef CK_RV (*CK_C_Logout) (CK_SESSION_HANDLE session);
-typedef CK_RV (*CK_C_CreateObject) (CK_SESSION_HANDLE session,
- struct _CK_ATTRIBUTE *templ,
- unsigned long count,
- CK_OBJECT_HANDLE *object);
-typedef ... *CK_C_CopyObject;
-typedef CK_RV (*CK_C_DestroyObject) (CK_SESSION_HANDLE session,
- CK_OBJECT_HANDLE object);
-typedef ... *CK_C_GetObjectSize;
-typedef CK_RV (*CK_C_GetAttributeValue) (CK_SESSION_HANDLE session,
- CK_OBJECT_HANDLE object,
- struct _CK_ATTRIBUTE *templ,
- unsigned long count);
-typedef CK_RV (*CK_C_SetAttributeValue) (CK_SESSION_HANDLE session,
- CK_OBJECT_HANDLE object,
- struct _CK_ATTRIBUTE *templ,
- unsigned long count);
-typedef CK_RV (*CK_C_FindObjectsInit) (CK_SESSION_HANDLE session,
- struct _CK_ATTRIBUTE *templ,
- unsigned long count);
-typedef CK_RV (*CK_C_FindObjects) (CK_SESSION_HANDLE session,
- CK_OBJECT_HANDLE *object,
- unsigned long max_object_count,
- unsigned long *object_count);
-typedef CK_RV (*CK_C_FindObjectsFinal) (CK_SESSION_HANDLE session);
-typedef ... *CK_C_EncryptInit;
-typedef ... *CK_C_Encrypt;
-typedef ... *CK_C_EncryptUpdate;
-typedef ... *CK_C_EncryptFinal;
-typedef ... *CK_C_DecryptInit;
-typedef ... *CK_C_Decrypt;
-typedef ... *CK_C_DecryptUpdate;
-typedef ... *CK_C_DecryptFinal;
-typedef ... *CK_C_DigestInit;
-typedef ... *CK_C_Digest;
-typedef ... *CK_C_DigestUpdate;
-typedef ... *CK_C_DigestKey;
-typedef ... *CK_C_DigestFinal;
-typedef ... *CK_C_SignInit;
-typedef ... *CK_C_Sign;
-typedef ... *CK_C_SignUpdate;
-typedef ... *CK_C_SignFinal;
-typedef ... *CK_C_SignRecoverInit;
-typedef ... *CK_C_SignRecover;
-typedef ... *CK_C_VerifyInit;
-typedef ... *CK_C_Verify;
-typedef ... *CK_C_VerifyUpdate;
-typedef ... *CK_C_VerifyFinal;
-typedef ... *CK_C_VerifyRecoverInit;
-typedef ... *CK_C_VerifyRecover;
-typedef ... *CK_C_DigestEncryptUpdate;
-typedef ... *CK_C_DecryptDigestUpdate;
-typedef ... *CK_C_SignEncryptUpdate;
-typedef ... *CK_C_DecryptVerifyUpdate;
-typedef CK_RV (*CK_C_GenerateKey) (CK_SESSION_HANDLE session,
- struct _CK_MECHANISM *mechanism,
- struct _CK_ATTRIBUTE *templ,
- unsigned long count,
- CK_OBJECT_HANDLE *key);
-typedef CK_RV (*CK_C_GenerateKeyPair) (CK_SESSION_HANDLE session,
- struct _CK_MECHANISM *mechanism,
- struct _CK_ATTRIBUTE *
- public_key_template,
- unsigned long
- public_key_attribute_count,
- struct _CK_ATTRIBUTE *
- private_key_template,
- unsigned long
- private_key_attribute_count,
- CK_OBJECT_HANDLE *public_key,
- CK_OBJECT_HANDLE *private_key);
-typedef CK_RV (*CK_C_WrapKey) (CK_SESSION_HANDLE session,
- struct _CK_MECHANISM *mechanism,
- CK_OBJECT_HANDLE wrapping_key,
- CK_OBJECT_HANDLE key,
- unsigned char *wrapped_key,
- unsigned long *wrapped_key_len);
-typedef CK_RV (*CK_C_UnwrapKey) (CK_SESSION_HANDLE session,
- struct _CK_MECHANISM *mechanism,
- CK_OBJECT_HANDLE unwrapping_key,
- unsigned char *wrapped_key,
- unsigned long wrapped_key_len,
- struct _CK_ATTRIBUTE *templ,
- unsigned long attribute_count,
- CK_OBJECT_HANDLE *key);
-typedef ... *CK_C_DeriveKey;
-typedef ... *CK_C_SeedRandom;
-typedef ... *CK_C_GenerateRandom;
-typedef ... *CK_C_GetFunctionStatus;
-typedef ... *CK_C_CancelFunction;
-
-struct _CK_FUNCTION_LIST
-{
- struct _CK_VERSION version;
- CK_C_Initialize C_Initialize;
- CK_C_Finalize C_Finalize;
- CK_C_GetInfo C_GetInfo;
- CK_C_GetFunctionList C_GetFunctionList;
- CK_C_GetSlotList C_GetSlotList;
- CK_C_GetSlotInfo C_GetSlotInfo;
- CK_C_GetTokenInfo C_GetTokenInfo;
- CK_C_GetMechanismList C_GetMechanismList;
- CK_C_GetMechanismInfo C_GetMechanismInfo;
- CK_C_InitToken C_InitToken;
- CK_C_InitPIN C_InitPIN;
- CK_C_SetPIN C_SetPIN;
- CK_C_OpenSession C_OpenSession;
- CK_C_CloseSession C_CloseSession;
- CK_C_CloseAllSessions C_CloseAllSessions;
- CK_C_GetSessionInfo C_GetSessionInfo;
- CK_C_GetOperationState C_GetOperationState;
- CK_C_SetOperationState C_SetOperationState;
- CK_C_Login C_Login;
- CK_C_Logout C_Logout;
- CK_C_CreateObject C_CreateObject;
- CK_C_CopyObject C_CopyObject;
- CK_C_DestroyObject C_DestroyObject;
- CK_C_GetObjectSize C_GetObjectSize;
- CK_C_GetAttributeValue C_GetAttributeValue;
- CK_C_SetAttributeValue C_SetAttributeValue;
- CK_C_FindObjectsInit C_FindObjectsInit;
- CK_C_FindObjects C_FindObjects;
- CK_C_FindObjectsFinal C_FindObjectsFinal;
- CK_C_EncryptInit C_EncryptInit;
- CK_C_Encrypt C_Encrypt;
- CK_C_EncryptUpdate C_EncryptUpdate;
- CK_C_EncryptFinal C_EncryptFinal;
- CK_C_DecryptInit C_DecryptInit;
- CK_C_Decrypt C_Decrypt;
- CK_C_DecryptUpdate C_DecryptUpdate;
- CK_C_DecryptFinal C_DecryptFinal;
- CK_C_DigestInit C_DigestInit;
- CK_C_Digest C_Digest;
- CK_C_DigestUpdate C_DigestUpdate;
- CK_C_DigestKey C_DigestKey;
- CK_C_DigestFinal C_DigestFinal;
- CK_C_SignInit C_SignInit;
- CK_C_Sign C_Sign;
- CK_C_SignUpdate C_SignUpdate;
- CK_C_SignFinal C_SignFinal;
- CK_C_SignRecoverInit C_SignRecoverInit;
- CK_C_SignRecover C_SignRecover;
- CK_C_VerifyInit C_VerifyInit;
- CK_C_Verify C_Verify;
- CK_C_VerifyUpdate C_VerifyUpdate;
- CK_C_VerifyFinal C_VerifyFinal;
- CK_C_VerifyRecoverInit C_VerifyRecoverInit;
- CK_C_VerifyRecover C_VerifyRecover;
- CK_C_DigestEncryptUpdate C_DigestEncryptUpdate;
- CK_C_DecryptDigestUpdate C_DecryptDigestUpdate;
- CK_C_SignEncryptUpdate C_SignEncryptUpdate;
- CK_C_DecryptVerifyUpdate C_DecryptVerifyUpdate;
- CK_C_GenerateKey C_GenerateKey;
- CK_C_GenerateKeyPair C_GenerateKeyPair;
- CK_C_WrapKey C_WrapKey;
- CK_C_UnwrapKey C_UnwrapKey;
- CK_C_DeriveKey C_DeriveKey;
- CK_C_SeedRandom C_SeedRandom;
- CK_C_GenerateRandom C_GenerateRandom;
- CK_C_GetFunctionStatus C_GetFunctionStatus;
- CK_C_CancelFunction C_CancelFunction;
- CK_C_WaitForSlotEvent C_WaitForSlotEvent;
-};
-
-typedef unsigned char CK_BYTE;
-typedef unsigned char CK_UTF8CHAR;
-typedef unsigned char CK_BBOOL;
-typedef unsigned long int CK_ULONG;
-typedef CK_BYTE *CK_BYTE_PTR;
-typedef CK_ULONG *CK_ULONG_PTR;
-
-typedef CK_OBJECT_HANDLE *CK_OBJECT_HANDLE_PTR;
-
-typedef struct _CK_ATTRIBUTE CK_ATTRIBUTE;
-typedef struct _CK_ATTRIBUTE *CK_ATTRIBUTE_PTR;
-
-typedef struct _CK_MECHANISM CK_MECHANISM;
-
-typedef struct _CK_FUNCTION_LIST *CK_FUNCTION_LIST_PTR;
-
-
-/* p11-kit/uri.h */
-
-typedef enum {
- DUMMY /* ..., */
-} P11KitUriType;
-
-typedef ... P11KitUri;
-
-CK_ATTRIBUTE_PTR p11_kit_uri_get_attributes (P11KitUri *uri,
- CK_ULONG *n_attrs);
-
-int p11_kit_uri_any_unrecognized (P11KitUri *uri);
-
-P11KitUri* p11_kit_uri_new (void);
-
-int p11_kit_uri_parse (const char *string,
- P11KitUriType uri_type,
- P11KitUri *uri);
-
-void p11_kit_uri_free (P11KitUri *uri);
-
-
-/* p11helper.c */
-
-struct ck_rsa_pkcs_oaep_params {
- CK_MECHANISM_TYPE hash_alg;
- unsigned long mgf;
- unsigned long source;
- void *source_data;
- unsigned long source_data_len;
-};
-
-typedef struct ck_rsa_pkcs_oaep_params CK_RSA_PKCS_OAEP_PARAMS;
-''')
-
-_libp11_kit = _ffi.dlopen(ctypes.util.find_library('p11-kit'))
-
-
-# utility
-
-NULL = _ffi.NULL
-
-unsigned_char = _ffi.typeof('unsigned char')
-unsigned_long = _ffi.typeof('unsigned long')
-
-sizeof = _ffi.sizeof
-
-
-def new_ptr(ctype, *args):
- return _ffi.new(_ffi.getctype(ctype, '*'), *args)
-
-
-def new_array(ctype, *args):
- return _ffi.new(_ffi.getctype(ctype, '[]'), *args)
-
-
-# p11-kit/pkcs11.h
-
-CK_SESSION_HANDLE = _ffi.typeof('CK_SESSION_HANDLE')
-
-CK_OBJECT_HANDLE = _ffi.typeof('CK_OBJECT_HANDLE')
-
-CKU_USER = 1
-
-CKF_RW_SESSION = 0x2
-CKF_SERIAL_SESSION = 0x4
-
-CK_OBJECT_CLASS = _ffi.typeof('CK_OBJECT_CLASS')
-
-CKO_PUBLIC_KEY = 2
-CKO_PRIVATE_KEY = 3
-CKO_SECRET_KEY = 4
-CKO_VENDOR_DEFINED = 0x80000000
-
-CK_KEY_TYPE = _ffi.typeof('CK_KEY_TYPE')
-
-CKK_RSA = 0
-CKK_AES = 0x1f
-
-CKA_CLASS = 0
-CKA_TOKEN = 1
-CKA_PRIVATE = 2
-CKA_LABEL = 3
-CKA_TRUSTED = 0x86
-CKA_KEY_TYPE = 0x100
-CKA_ID = 0x102
-CKA_SENSITIVE = 0x103
-CKA_ENCRYPT = 0x104
-CKA_DECRYPT = 0x105
-CKA_WRAP = 0x106
-CKA_UNWRAP = 0x107
-CKA_SIGN = 0x108
-CKA_SIGN_RECOVER = 0x109
-CKA_VERIFY = 0x10a
-CKA_VERIFY_RECOVER = 0x10b
-CKA_DERIVE = 0x10c
-CKA_MODULUS = 0x120
-CKA_MODULUS_BITS = 0x121
-CKA_PUBLIC_EXPONENT = 0x122
-CKA_VALUE_LEN = 0x161
-CKA_EXTRACTABLE = 0x162
-CKA_LOCAL = 0x163
-CKA_NEVER_EXTRACTABLE = 0x164
-CKA_ALWAYS_SENSITIVE = 0x165
-CKA_MODIFIABLE = 0x170
-CKA_ALWAYS_AUTHENTICATE = 0x202
-CKA_WRAP_WITH_TRUSTED = 0x210
-
-CKM_RSA_PKCS_KEY_PAIR_GEN = 0
-CKM_RSA_PKCS = 1
-CKM_RSA_PKCS_OAEP = 9
-CKM_SHA_1 = 0x220
-CKM_AES_KEY_GEN = 0x1080
-
-CKR_OK = 0
-CKR_ATTRIBUTE_TYPE_INVALID = 0x12
-CKR_USER_NOT_LOGGED_IN = 0x101
-
-CK_BYTE = _ffi.typeof('CK_BYTE')
-CK_BBOOL = _ffi.typeof('CK_BBOOL')
-CK_ULONG = _ffi.typeof('CK_ULONG')
-CK_BYTE_PTR = _ffi.typeof('CK_BYTE_PTR')
-CK_FALSE = 0
-CK_TRUE = 1
-
-CK_OBJECT_HANDLE_PTR = _ffi.typeof('CK_OBJECT_HANDLE_PTR')
-
-CK_ATTRIBUTE = _ffi.typeof('CK_ATTRIBUTE')
-
-CK_MECHANISM = _ffi.typeof('CK_MECHANISM')
-
-CK_FUNCTION_LIST_PTR = _ffi.typeof('CK_FUNCTION_LIST_PTR')
-
-NULL_PTR = NULL
-
-
-# p11-kit/uri.h
-
-P11_KIT_URI_OK = 0
-
-P11_KIT_URI_FOR_OBJECT = 2
-
-p11_kit_uri_get_attributes = _libp11_kit.p11_kit_uri_get_attributes
-
-p11_kit_uri_any_unrecognized = _libp11_kit.p11_kit_uri_any_unrecognized
-
-p11_kit_uri_new = _libp11_kit.p11_kit_uri_new
-
-p11_kit_uri_parse = _libp11_kit.p11_kit_uri_parse
-
-p11_kit_uri_free = _libp11_kit.p11_kit_uri_free
-
-
-# library.c
-
-def loadLibrary(module):
- """Load the PKCS#11 library"""
- # Load PKCS #11 library
- try:
- if module:
- # pylint: disable=no-member
- pDynLib = _ffi.dlopen(module, _ffi.RTLD_NOW | _ffi.RTLD_LOCAL)
- else:
- raise Exception()
-
- except Exception:
- # Failed to load the PKCS #11 library
- raise
-
- # Retrieve the entry point for C_GetFunctionList
- pGetFunctionList = pDynLib.C_GetFunctionList
- if pGetFunctionList == NULL:
- raise Exception()
-
- # Store the handle so we can dlclose it later
-
- return pGetFunctionList, pDynLib
-
-
-# p11helper.c
-
-# compat TODO
-CKM_AES_KEY_WRAP = 0x2109
-CKM_AES_KEY_WRAP_PAD = 0x210a
-
-# TODO
-CKA_COPYABLE = 0x0017
-
-CKG_MGF1_SHA1 = 0x00000001
-
-CKZ_DATA_SPECIFIED = 0x00000001
-
-CK_RSA_PKCS_OAEP_PARAMS = _ffi.typeof('CK_RSA_PKCS_OAEP_PARAMS')
-
-
-true_ptr = new_ptr(CK_BBOOL, CK_TRUE)
-false_ptr = new_ptr(CK_BBOOL, CK_FALSE)
-
-MAX_TEMPLATE_LEN = 32
-
-#
-# Constants
-#
-CONST_RSA_PKCS_OAEP_PARAMS_ptr = new_ptr(CK_RSA_PKCS_OAEP_PARAMS, dict(
- hash_alg=CKM_SHA_1,
- mgf=CKG_MGF1_SHA1,
- source=CKZ_DATA_SPECIFIED,
- source_data=NULL,
- source_data_len=0,
-))
-
-
-#
-# ipap11helper Exceptions
-#
-class P11HelperException(Exception):
- """parent class for all exceptions"""
- pass
-P11HelperException.__name__ = 'Exception'
-
-
-class Error(P11HelperException):
- """general error"""
- pass
-
-
-class NotFound(P11HelperException):
- """key not found"""
- pass
-
-
-class DuplicationError(P11HelperException):
- """key already exists"""
- pass
-
-
-########################################################################
-# Support functions
-#
-
-def pyobj_to_bool(pyobj):
- if pyobj:
- return true_ptr
- return false_ptr
-
-
-def convert_py2bool(mapping):
- return tuple(pyobj_to_bool(py_obj) for py_obj in mapping)
-
-
-def string_to_pybytes_or_none(str, len):
- if str == NULL:
- return None
- return _ffi.buffer(str, len)[:]
-
-
-def unicode_to_char_array(unicode):
- """
- Convert a unicode string to the utf8 encoded char array
- :param unicode: input python unicode object
- """
- try:
- utf8_str = unicode.encode('utf-8')
- except Exception:
- raise Error("Unable to encode UTF-8")
- try:
- result = new_array(unsigned_char, utf8_str)
- except Exception:
- raise Error("Unable to get bytes from string")
- l = len(utf8_str)
- return result, l
-
-
-def char_array_to_unicode(array, l):
- """
- Convert utf-8 encoded char array to unicode object
- """
- return _ffi.buffer(array, l)[:].decode('utf-8')
-
-
-def int_to_bytes(value):
- try:
- return binascii.unhexlify('{0:x}'.format(value))
- except (TypeError, binascii.Error):
- return binascii.unhexlify('0{0:x}'.format(value))
-
-
-def bytes_to_int(value):
- return int(binascii.hexlify(value), 16)
-
-
-def check_return_value(rv, message):
- """
- Tests result value of pkc11 operations
- """
- if rv != CKR_OK:
- try:
- errmsg = "Error at %s: 0x%x\n" % (message, rv)
- except Exception:
- raise Error("An error occured during error message generation. "
- "Please report this problem. Developers will use "
- "a crystal ball to find out the root cause.")
- else:
- raise Error(errmsg)
-
-
-def _fill_template_from_parts(attr, template_len, id, id_len, label, label_len,
- class_, cka_wrap, cka_unwrap):
- """
- Fill template structure with pointers to attributes passed as independent
- variables.
- Variables with NULL values will be omitted from template.
-
- @warning input variables should not be modified when template is in use
- """
- cnt = 0
- if label != NULL:
- attr[0].type = CKA_LABEL
- attr[0].pValue = label
- attr[0].ulValueLen = label_len
- attr += 1
- cnt += 1
- assert cnt < template_len[0]
- if id != NULL:
- attr[0].type = CKA_ID
- attr[0].pValue = id
- attr[0].ulValueLen = id_len
- attr += 1
- cnt += 1
- assert cnt < template_len[0]
- if cka_wrap != NULL:
- attr[0].type = CKA_WRAP
- attr[0].pValue = cka_wrap
- attr[0].ulValueLen = sizeof(CK_BBOOL)
- attr += 1
- cnt += 1
- assert cnt < template_len[0]
- if cka_unwrap != NULL:
- attr[0].type = CKA_UNWRAP
- attr[0].pValue = cka_unwrap
- attr[0].ulValueLen = sizeof(CK_BBOOL)
- attr += 1
- cnt += 1
- assert cnt < template_len[0]
-
- if class_ != NULL:
- attr[0].type = CKA_CLASS
- attr[0].pValue = class_
- attr[0].ulValueLen = sizeof(CK_OBJECT_CLASS)
- attr += 1
- cnt += 1
- assert cnt < template_len[0]
- template_len[0] = cnt
-
-
-def _parse_uri(uri_str):
- """
- Parse string to P11-kit representation of PKCS#11 URI.
- """
- uri = p11_kit_uri_new()
- if not uri:
- raise Error("Cannot initialize URI parser")
-
- try:
- result = p11_kit_uri_parse(uri_str, P11_KIT_URI_FOR_OBJECT, uri)
- if result != P11_KIT_URI_OK:
- raise Error("Cannot parse URI")
-
- if p11_kit_uri_any_unrecognized(uri):
- raise Error("PKCS#11 URI contains unsupported attributes")
- except Error:
- p11_kit_uri_free(uri)
- raise
-
- return uri
-
-
-def _set_wrapping_mech_parameters(mech_type, mech):
- """
- Function set default param values for wrapping mechanism
- :param mech_type: mechanism type
- :param mech: filled structure with params based on mech type
-
- Warning: do not dealloc param values, it is static variables
- """
- if mech_type in (CKM_RSA_PKCS, CKM_AES_KEY_WRAP, CKM_AES_KEY_WRAP_PAD):
- mech.pParameter = NULL
- mech.ulParameterLen = 0
- elif mech_type == CKM_RSA_PKCS_OAEP:
- # Use the same configuration as openSSL
- # https://www.openssl.org/docs/crypto/RSA_public_encrypt.html
- mech.pParameter = CONST_RSA_PKCS_OAEP_PARAMS_ptr
- mech.ulParameterLen = sizeof(CK_RSA_PKCS_OAEP_PARAMS)
- else:
- raise Error("Unsupported wrapping mechanism")
- mech.mechanism = mech_type
-
-
-########################################################################
-# P11_Helper object
-#
-class P11_Helper(object):
- @property
- def p11(self):
- return self.p11_ptr[0]
-
- @property
- def session(self):
- return self.session_ptr[0]
-
- def _find_key(self, template, template_len):
- """
- Find keys matching specified template.
- Function returns list of key handles via objects parameter.
-
- :param template: PKCS#11 template for attribute matching
- """
- result_objects = []
- result_object_ptr = new_ptr(CK_OBJECT_HANDLE)
- objectCount_ptr = new_ptr(CK_ULONG)
-
- rv = self.p11.C_FindObjectsInit(self.session, template, template_len)
- check_return_value(rv, "Find key init")
-
- rv = self.p11.C_FindObjects(self.session, result_object_ptr, 1,
- objectCount_ptr)
- check_return_value(rv, "Find key")
-
- while objectCount_ptr[0] > 0:
- result_objects.append(result_object_ptr[0])
-
- rv = self.p11.C_FindObjects(self.session, result_object_ptr, 1,
- objectCount_ptr)
- check_return_value(rv, "Check for duplicated key")
-
- rv = self.p11.C_FindObjectsFinal(self.session)
- check_return_value(rv, "Find objects final")
-
- return result_objects
-
- def _id_exists(self, id, id_len, class_):
- """
- Test if object with specified label, id and class exists
-
- :param id: key ID, (if value is NULL, will not be used to find key)
- :param id_len: key ID length
- :param class_ key: class
-
- :return: True if object was found, False if object doesnt exists
- """
- object_count_ptr = new_ptr(CK_ULONG)
- result_object_ptr = new_ptr(CK_OBJECT_HANDLE)
- class_ptr = new_ptr(CK_OBJECT_CLASS, class_)
- class_sec_ptr = new_ptr(CK_OBJECT_CLASS, CKO_SECRET_KEY)
-
- template_pub_priv = new_array(CK_ATTRIBUTE, (
- (CKA_ID, id, id_len),
- (CKA_CLASS, class_ptr, sizeof(CK_OBJECT_CLASS)),
- ))
-
- template_sec = new_array(CK_ATTRIBUTE, (
- (CKA_ID, id, id_len),
- (CKA_CLASS, class_sec_ptr, sizeof(CK_OBJECT_CLASS)),
- ))
-
- template_id = new_array(CK_ATTRIBUTE, (
- (CKA_ID, id, id_len),
- ))
-
- #
- # Only one secret key with same ID is allowed
- #
- if class_ == CKO_SECRET_KEY:
- rv = self.p11.C_FindObjectsInit(self.session, template_id, 1)
- check_return_value(rv, "id, label exists init")
-
- rv = self.p11.C_FindObjects(self.session, result_object_ptr, 1,
- object_count_ptr)
- check_return_value(rv, "id, label exists")
-
- rv = self.p11.C_FindObjectsFinal(self.session)
- check_return_value(rv, "id, label exists final")
-
- if object_count_ptr[0] > 0:
- return True
- return False
-
- #
- # Public and private keys can share one ID, but
- #
-
- # test if secret key with same ID exists
- rv = self.p11.C_FindObjectsInit(self.session, template_sec, 2)
- check_return_value(rv, "id, label exists init")
-
- rv = self.p11.C_FindObjects(self.session, result_object_ptr, 1,
- object_count_ptr)
- check_return_value(rv, "id, label exists")
-
- rv = self.p11.C_FindObjectsFinal(self.session)
- check_return_value(rv, "id, label exists final")
-
- if object_count_ptr[0] > 0:
- # object found
- return True
-
- # test if pub/private key with same id exists
- object_count_ptr[0] = 0
-
- rv = self.p11.C_FindObjectsInit(self.session, template_pub_priv, 2)
- check_return_value(rv, "id, label exists init")
-
- rv = self.p11.C_FindObjects(self.session, result_object_ptr, 1,
- object_count_ptr)
- check_return_value(rv, "id, label exists")
-
- rv = self.p11.C_FindObjectsFinal(self.session)
- check_return_value(rv, "id, label exists final")
-
- if object_count_ptr[0] > 0:
- # Object found
- return True
-
- # Object not found
- return False
-
- def __init__(self, slot, user_pin, library_path):
- self.p11_ptr = new_ptr(CK_FUNCTION_LIST_PTR)
- self.session_ptr = new_ptr(CK_SESSION_HANDLE)
-
- self.slot = 0
- self.session_ptr[0] = 0
- self.p11_ptr[0] = NULL
- self.module_handle = None
-
- # Parse method args
- if isinstance(user_pin, unicode):
- user_pin = user_pin.encode()
- self.slot = slot
-
- try:
- pGetFunctionList, module_handle = loadLibrary(library_path)
- except Exception:
- raise Error("Could not load the library.")
-
- self.module_handle = module_handle
-
- #
- # Load the function list
- #
- pGetFunctionList(self.p11_ptr)
-
- #
- # Initialize
- #
- rv = self.p11.C_Initialize(NULL)
- check_return_value(rv, "initialize")
-
- #
- # Start session
- #
- rv = self.p11.C_OpenSession(self.slot,
- CKF_SERIAL_SESSION | CKF_RW_SESSION, NULL,
- NULL, self.session_ptr)
- check_return_value(rv, "open session")
-
- #
- # Login
- #
- rv = self.p11.C_Login(self.session, CKU_USER, user_pin, len(user_pin))
- check_return_value(rv, "log in")
-
- def finalize(self):
- """
- Finalize operations with pkcs11 library
- """
- if self.p11 == NULL:
- return
-
- #
- # Logout
- #
- rv = self.p11.C_Logout(self.session)
- check_return_value(rv, "log out")
-
- #
- # End session
- #
- rv = self.p11.C_CloseSession(self.session)
- check_return_value(rv, "close session")
-
- #
- # Finalize
- #
- self.p11.C_Finalize(NULL)
-
- self.p11_ptr[0] = NULL
- self.session_ptr[0] = 0
- self.slot = 0
- self.module_handle = None
-
- #################################################################
- # Methods working with keys
- #
-
- def generate_master_key(self, label, id, key_length=16, cka_copyable=True,
- cka_decrypt=False, cka_derive=False,
- cka_encrypt=False, cka_extractable=True,
- cka_modifiable=True, cka_private=True,
- cka_sensitive=True, cka_sign=False,
- cka_unwrap=True, cka_verify=False, cka_wrap=True,
- cka_wrap_with_trusted=False):
- """
- Generate master key
-
- :return: master key handle
- """
- if isinstance(id, unicode):
- id = id.encode()
-
- attrs = (
- cka_copyable,
- cka_decrypt,
- cka_derive,
- cka_encrypt,
- cka_extractable,
- cka_modifiable,
- cka_private,
- cka_sensitive,
- cka_sign,
- cka_unwrap,
- cka_verify,
- cka_wrap,
- cka_wrap_with_trusted,
- )
-
- key_length_ptr = new_ptr(CK_ULONG, key_length)
- master_key_ptr = new_ptr(CK_OBJECT_HANDLE)
-
- label_unicode = label
- id_length = len(id)
- id_ = new_array(CK_BYTE, id)
- # TODO check long overflow
-
- label, label_length = unicode_to_char_array(label_unicode)
-
- # TODO param?
- mechanism_ptr = new_ptr(CK_MECHANISM, (
- CKM_AES_KEY_GEN, NULL_PTR, 0
- ))
-
- if key_length not in (16, 24, 32):
- raise Error("generate_master_key: key length allowed values are: "
- "16, 24 and 32")
-
- if self._id_exists(id_, id_length, CKO_SECRET_KEY):
- raise DuplicationError("Master key with same ID already exists")
-
- # Process keyword boolean arguments
- (_cka_copyable_ptr, cka_decrypt_ptr, cka_derive_ptr, cka_encrypt_ptr,
- cka_extractable_ptr, cka_modifiable_ptr, cka_private_ptr,
- cka_sensitive_ptr, cka_sign_ptr, cka_unwrap_ptr, cka_verify_ptr,
- cka_wrap_ptr, cka_wrap_with_trusted_ptr,) = convert_py2bool(attrs)
-
- symKeyTemplate = new_array(CK_ATTRIBUTE, (
- (CKA_ID, id_, id_length),
- (CKA_LABEL, label, label_length),
- (CKA_TOKEN, true_ptr, sizeof(CK_BBOOL)),
- (CKA_VALUE_LEN, key_length_ptr, sizeof(CK_ULONG)),
- # TODO Softhsm doesn't support it
- # (CKA_COPYABLE, cka_copyable_ptr, sizeof(CK_BBOOL)),
- (CKA_DECRYPT, cka_decrypt_ptr, sizeof(CK_BBOOL)),
- (CKA_DERIVE, cka_derive_ptr, sizeof(CK_BBOOL)),
- (CKA_ENCRYPT, cka_encrypt_ptr, sizeof(CK_BBOOL)),
- (CKA_EXTRACTABLE, cka_extractable_ptr, sizeof(CK_BBOOL)),
- (CKA_MODIFIABLE, cka_modifiable_ptr, sizeof(CK_BBOOL)),
- (CKA_PRIVATE, cka_private_ptr, sizeof(CK_BBOOL)),
- (CKA_SENSITIVE, cka_sensitive_ptr, sizeof(CK_BBOOL)),
- (CKA_SIGN, cka_sign_ptr, sizeof(CK_BBOOL)),
- (CKA_UNWRAP, cka_unwrap_ptr, sizeof(CK_BBOOL)),
- (CKA_VERIFY, cka_verify_ptr, sizeof(CK_BBOOL)),
- (CKA_WRAP, cka_wrap_ptr, sizeof(CK_BBOOL)),
- (CKA_WRAP_WITH_TRUSTED, cka_wrap_with_trusted_ptr,
- sizeof(CK_BBOOL)),
- ))
-
- rv = self.p11.C_GenerateKey(self.session, mechanism_ptr,
- symKeyTemplate,
- (sizeof(symKeyTemplate) //
- sizeof(CK_ATTRIBUTE)), master_key_ptr)
- check_return_value(rv, "generate master key")
-
- return master_key_ptr[0]
-
- def generate_replica_key_pair(self, label, id, modulus_bits=2048,
- pub_cka_copyable=True, pub_cka_derive=False,
- pub_cka_encrypt=False,
- pub_cka_modifiable=True,
- pub_cka_private=True, pub_cka_trusted=False,
- pub_cka_verify=False,
- pub_cka_verify_recover=False,
- pub_cka_wrap=True,
- priv_cka_always_authenticate=False,
- priv_cka_copyable=True,
- priv_cka_decrypt=False,
- priv_cka_derive=False,
- priv_cka_extractable=False,
- priv_cka_modifiable=True,
- priv_cka_private=True,
- priv_cka_sensitive=True,
- priv_cka_sign=False,
- priv_cka_sign_recover=False,
- priv_cka_unwrap=True,
- priv_cka_wrap_with_trusted=False):
- """
- Generate replica keys
-
- :returns: tuple (public_key_handle, private_key_handle)
- """
- if isinstance(id, unicode):
- id = id.encode()
-
- attrs_pub = (
- pub_cka_copyable,
- pub_cka_derive,
- pub_cka_encrypt,
- pub_cka_modifiable,
- pub_cka_private,
- pub_cka_trusted,
- pub_cka_verify,
- pub_cka_verify_recover,
- pub_cka_wrap,
- )
-
- attrs_priv = (
- priv_cka_always_authenticate,
- priv_cka_copyable,
- priv_cka_decrypt,
- priv_cka_derive,
- priv_cka_extractable,
- priv_cka_modifiable,
- priv_cka_private,
- priv_cka_sensitive,
- priv_cka_sign,
- priv_cka_sign_recover,
- priv_cka_unwrap,
- priv_cka_wrap_with_trusted,
- )
-
- label_unicode = label
- id_ = new_array(CK_BYTE, id)
- id_length = len(id)
-
- label, label_length = unicode_to_char_array(label_unicode)
-
- public_key_ptr = new_ptr(CK_OBJECT_HANDLE)
- private_key_ptr = new_ptr(CK_OBJECT_HANDLE)
- mechanism_ptr = new_ptr(CK_MECHANISM,
- (CKM_RSA_PKCS_KEY_PAIR_GEN, NULL_PTR, 0))
-
- if self._id_exists(id_, id_length, CKO_PRIVATE_KEY):
- raise DuplicationError("Private key with same ID already exists")
-
- if self._id_exists(id_, id_length, CKO_PUBLIC_KEY):
- raise DuplicationError("Public key with same ID already exists")
-
- modulus_bits_ptr = new_ptr(CK_ULONG, modulus_bits)
-
- # Process keyword boolean arguments
- (_pub_cka_copyable_ptr, pub_cka_derive_ptr, pub_cka_encrypt_ptr,
- pub_cka_modifiable_ptr, pub_cka_private_ptr, pub_cka_trusted_ptr,
- pub_cka_verify_ptr, pub_cka_verify_recover_ptr, pub_cka_wrap_ptr,
- ) = convert_py2bool(attrs_pub)
- (priv_cka_always_authenticate_ptr, _priv_cka_copyable_ptr,
- priv_cka_decrypt_ptr, priv_cka_derive_ptr, priv_cka_extractable_ptr,
- priv_cka_modifiable_ptr, priv_cka_private_ptr, priv_cka_sensitive_ptr,
- priv_cka_sign_ptr, _priv_cka_sign_recover_ptr, priv_cka_unwrap_ptr,
- priv_cka_wrap_with_trusted_ptr,) = convert_py2bool(attrs_priv)
-
- # 65537 (RFC 6376 section 3.3.1)
- public_exponent = new_array(CK_BYTE, (1, 0, 1))
- publicKeyTemplate = new_array(CK_ATTRIBUTE, (
- (CKA_ID, id_, id_length),
- (CKA_LABEL, label, label_length),
- (CKA_TOKEN, true_ptr, sizeof(CK_BBOOL)),
- (CKA_MODULUS_BITS, modulus_bits_ptr, sizeof(CK_ULONG)),
- (CKA_PUBLIC_EXPONENT, public_exponent, 3),
- # TODO Softhsm doesn't support it
- # (CKA_COPYABLE, pub_cka_copyable_p, sizeof(CK_BBOOL)),
- (CKA_DERIVE, pub_cka_derive_ptr, sizeof(CK_BBOOL)),
- (CKA_ENCRYPT, pub_cka_encrypt_ptr, sizeof(CK_BBOOL)),
- (CKA_MODIFIABLE, pub_cka_modifiable_ptr, sizeof(CK_BBOOL)),
- (CKA_PRIVATE, pub_cka_private_ptr, sizeof(CK_BBOOL)),
- (CKA_TRUSTED, pub_cka_trusted_ptr, sizeof(CK_BBOOL)),
- (CKA_VERIFY, pub_cka_verify_ptr, sizeof(CK_BBOOL)),
- (CKA_VERIFY_RECOVER, pub_cka_verify_recover_ptr, sizeof(CK_BBOOL)),
- (CKA_WRAP, pub_cka_wrap_ptr, sizeof(CK_BBOOL)),
- ))
-
- privateKeyTemplate = new_array(CK_ATTRIBUTE, (
- (CKA_ID, id_, id_length),
- (CKA_LABEL, label, label_length),
- (CKA_TOKEN, true_ptr, sizeof(CK_BBOOL)),
- (CKA_ALWAYS_AUTHENTICATE, priv_cka_always_authenticate_ptr,
- sizeof(CK_BBOOL)),
- # TODO Softhsm doesn't support it
- # (CKA_COPYABLE, priv_cka_copyable_ptr, sizeof(CK_BBOOL)),
- (CKA_DECRYPT, priv_cka_decrypt_ptr, sizeof(CK_BBOOL)),
- (CKA_DERIVE, priv_cka_derive_ptr, sizeof(CK_BBOOL)),
- (CKA_EXTRACTABLE, priv_cka_extractable_ptr, sizeof(CK_BBOOL)),
- (CKA_MODIFIABLE, priv_cka_modifiable_ptr, sizeof(CK_BBOOL)),
- (CKA_PRIVATE, priv_cka_private_ptr, sizeof(CK_BBOOL)),
- (CKA_SENSITIVE, priv_cka_sensitive_ptr, sizeof(CK_BBOOL)),
- (CKA_SIGN, priv_cka_sign_ptr, sizeof(CK_BBOOL)),
- (CKA_SIGN_RECOVER, priv_cka_sign_ptr, sizeof(CK_BBOOL)),
- (CKA_UNWRAP, priv_cka_unwrap_ptr, sizeof(CK_BBOOL)),
- (CKA_WRAP_WITH_TRUSTED, priv_cka_wrap_with_trusted_ptr,
- sizeof(CK_BBOOL)),
- ))
-
- rv = self.p11.C_GenerateKeyPair(self.session, mechanism_ptr,
- publicKeyTemplate,
- (sizeof(publicKeyTemplate) //
- sizeof(CK_ATTRIBUTE)),
- privateKeyTemplate,
- (sizeof(privateKeyTemplate) //
- sizeof(CK_ATTRIBUTE)),
- public_key_ptr,
- private_key_ptr)
- check_return_value(rv, "generate key pair")
-
- return public_key_ptr[0], private_key_ptr[0]
-
- def find_keys(self, objclass=CKO_VENDOR_DEFINED, label=None, id=None,
- cka_wrap=None, cka_unwrap=None, uri=None):
- """
- Find key
- """
- if isinstance(id, unicode):
- id = id.encode()
- if isinstance(uri, unicode):
- uri = uri.encode()
-
- class_ = objclass
- class_ptr = new_ptr(CK_OBJECT_CLASS, class_)
- ckawrap = NULL
- ckaunwrap = NULL
- if id is not None:
- id_ = new_array(CK_BYTE, id)
- id_length = len(id)
- else:
- id_ = NULL
- id_length = 0
- label_unicode, label = label, NULL
- cka_wrap_bool = cka_wrap
- cka_unwrap_bool = cka_unwrap
- label_length = 0
- uri_str = uri
- uri = NULL
- template = new_array(CK_ATTRIBUTE, MAX_TEMPLATE_LEN)
- template_len_ptr = new_ptr(CK_ULONG, MAX_TEMPLATE_LEN)
-
- # TODO check long overflow
-
- if label_unicode is not None:
- label, label_length = unicode_to_char_array(label_unicode)
-
- if cka_wrap_bool is not None:
- if cka_wrap_bool:
- ckawrap = true_ptr
- else:
- ckawrap = false_ptr
-
- if cka_unwrap_bool is not None:
- if cka_unwrap_bool:
- ckaunwrap = true_ptr
- else:
- ckaunwrap = false_ptr
-
- if class_ == CKO_VENDOR_DEFINED:
- class_ptr = NULL
-
- try:
- if uri_str is None:
- _fill_template_from_parts(template, template_len_ptr, id_,
- id_length, label, label_length,
- class_ptr, ckawrap, ckaunwrap)
- else:
- uri = _parse_uri(uri_str)
- template = (p11_kit_uri_get_attributes(uri, template_len_ptr))
- # Do not deallocate URI while you are using the template.
- # Template contains pointers to values inside URI!
-
- result_list = self._find_key(template, template_len_ptr[0])
-
- return result_list
- finally:
- if uri != NULL:
- p11_kit_uri_free(uri)
-
- def delete_key(self, key_handle):
- """
- delete key
- """
- # TODO check long overflow
- rv = self.p11.C_DestroyObject(self.session, key_handle)
- check_return_value(rv, "object deletion")
-
- def _export_RSA_public_key(self, object):
- """
- export RSA public key
- """
- class_ptr = new_ptr(CK_OBJECT_CLASS, CKO_PUBLIC_KEY)
- key_type_ptr = new_ptr(CK_KEY_TYPE, CKK_RSA)
-
- obj_template = new_array(CK_ATTRIBUTE, (
- (CKA_MODULUS, NULL_PTR, 0),
- (CKA_PUBLIC_EXPONENT, NULL_PTR, 0),
- (CKA_CLASS, class_ptr, sizeof(CK_OBJECT_CLASS)),
- (CKA_KEY_TYPE, key_type_ptr, sizeof(CK_KEY_TYPE)),
- ))
-
- rv = self.p11.C_GetAttributeValue(self.session, object, obj_template,
- (sizeof(obj_template) //
- sizeof(CK_ATTRIBUTE)))
- check_return_value(rv, "get RSA public key values - prepare")
-
- # Set proper size for attributes
- modulus = new_array(CK_BYTE,
- obj_template[0].ulValueLen * sizeof(CK_BYTE))
- obj_template[0].pValue = modulus
- exponent = new_array(CK_BYTE,
- obj_template[1].ulValueLen * sizeof(CK_BYTE))
- obj_template[1].pValue = exponent
-
- rv = self.p11.C_GetAttributeValue(self.session, object, obj_template,
- (sizeof(obj_template) //
- sizeof(CK_ATTRIBUTE)))
- check_return_value(rv, "get RSA public key values")
-
- # Check if the key is RSA public key
- if class_ptr[0] != CKO_PUBLIC_KEY:
- raise Error("export_RSA_public_key: required public key class")
-
- if key_type_ptr[0] != CKK_RSA:
- raise Error("export_RSA_public_key: required RSA key type")
-
- try:
- n = bytes_to_int(string_to_pybytes_or_none(
- modulus, obj_template[0].ulValueLen))
- except Exception:
- raise Error("export_RSA_public_key: internal error: unable to "
- "convert modulus")
-
- try:
- e = bytes_to_int(string_to_pybytes_or_none(
- exponent, obj_template[1].ulValueLen))
- except Exception:
- raise Error("export_RSA_public_key: internal error: unable to "
- "convert exponent")
-
- # set modulus and exponent
- rsa_ = rsa.RSAPublicNumbers(e, n)
-
- try:
- pkey = rsa_.public_key(default_backend())
- except Exception:
- raise Error("export_RSA_public_key: internal error: "
- "EVP_PKEY_set1_RSA failed")
-
- try:
- ret = pkey.public_bytes(
- format=serialization.PublicFormat.SubjectPublicKeyInfo,
- encoding=serialization.Encoding.DER,
- )
- except Exception:
- ret = None
-
- return ret
-
- def export_public_key(self, key_handle):
- """
- Export public key
-
- Export public key in SubjectPublicKeyInfo (RFC5280) DER encoded format
- """
- object = key_handle
- class_ptr = new_ptr(CK_OBJECT_CLASS, CKO_PUBLIC_KEY)
- key_type_ptr = new_ptr(CK_KEY_TYPE, CKK_RSA)
- # TODO check long overflow
-
- obj_template = new_array(CK_ATTRIBUTE, (
- (CKA_CLASS, class_ptr, sizeof(CK_OBJECT_CLASS)),
- (CKA_KEY_TYPE, key_type_ptr, sizeof(CK_KEY_TYPE)),
- ))
-
- rv = self.p11.C_GetAttributeValue(self.session, object, obj_template,
- (sizeof(obj_template) //
- sizeof(CK_ATTRIBUTE)))
- check_return_value(rv, "export_public_key: get RSA public key values")
-
- if class_ptr[0] != CKO_PUBLIC_KEY:
- raise Error("export_public_key: required public key class")
-
- if key_type_ptr[0] == CKK_RSA:
- return self._export_RSA_public_key(object)
- else:
- raise Error("export_public_key: unsupported key type")
-
- def _import_RSA_public_key(self, label, label_length, id, id_length, pkey,
- cka_copyable, cka_derive, cka_encrypt,
- cka_modifiable, cka_private, cka_trusted,
- cka_verify, cka_verify_recover, cka_wrap):
- """
- Import RSA public key
- """
- class_ptr = new_ptr(CK_OBJECT_CLASS, CKO_PUBLIC_KEY)
- keyType_ptr = new_ptr(CK_KEY_TYPE, CKK_RSA)
- cka_token = true_ptr
-
- if not isinstance(pkey, rsa.RSAPublicKey):
- raise Error("Required RSA public key")
-
- rsa_ = pkey.public_numbers()
-
- # convert BIGNUM to binary array
- modulus = new_array(CK_BYTE, int_to_bytes(rsa_.n))
- modulus_len = sizeof(modulus) - 1
- if modulus_len == 0:
- raise Error("import_RSA_public_key: BN_bn2bin modulus error")
-
- exponent = new_array(CK_BYTE, int_to_bytes(rsa_.e))
- exponent_len = sizeof(exponent) - 1
- if exponent_len == 0:
- raise Error("import_RSA_public_key: BN_bn2bin exponent error")
-
- template = new_array(CK_ATTRIBUTE, (
- (CKA_ID, id, id_length),
- (CKA_CLASS, class_ptr, sizeof(CK_OBJECT_CLASS)),
- (CKA_KEY_TYPE, keyType_ptr, sizeof(CK_KEY_TYPE)),
- (CKA_TOKEN, cka_token, sizeof(CK_BBOOL)),
- (CKA_LABEL, label, label_length),
- (CKA_MODULUS, modulus, modulus_len),
- (CKA_PUBLIC_EXPONENT, exponent, exponent_len),
- # TODO Softhsm doesn't support it
- # (CKA_COPYABLE, cka_copyable, sizeof(CK_BBOOL)),
- (CKA_DERIVE, cka_derive, sizeof(CK_BBOOL)),
- (CKA_ENCRYPT, cka_encrypt, sizeof(CK_BBOOL)),
- (CKA_MODIFIABLE, cka_modifiable, sizeof(CK_BBOOL)),
- (CKA_PRIVATE, cka_private, sizeof(CK_BBOOL)),
- (CKA_TRUSTED, cka_trusted, sizeof(CK_BBOOL)),
- (CKA_VERIFY, cka_verify, sizeof(CK_BBOOL)),
- (CKA_VERIFY_RECOVER, cka_verify_recover, sizeof(CK_BBOOL)),
- (CKA_WRAP, cka_wrap, sizeof(CK_BBOOL)),
- ))
- object_ptr = new_ptr(CK_OBJECT_HANDLE)
-
- rv = self.p11.C_CreateObject(self.session, template,
- (sizeof(template) //
- sizeof(CK_ATTRIBUTE)), object_ptr)
- check_return_value(rv, "create public key object")
-
- return object_ptr[0]
-
- def import_public_key(self, label, id, data, cka_copyable=True,
- cka_derive=False, cka_encrypt=False,
- cka_modifiable=True, cka_private=True,
- cka_trusted=False, cka_verify=True,
- cka_verify_recover=True, cka_wrap=False):
- """
- Import RSA public key
- """
- if isinstance(id, unicode):
- id = id.encode()
- if isinstance(data, unicode):
- data = data.encode()
-
- label_unicode = label
- id_ = new_array(CK_BYTE, id)
- id_length = len(id)
-
- attrs_pub = (
- cka_copyable,
- cka_derive,
- cka_encrypt,
- cka_modifiable,
- cka_private,
- cka_trusted,
- cka_verify,
- cka_verify_recover,
- cka_wrap,
- )
-
- label, label_length = unicode_to_char_array(label_unicode)
-
- if self._id_exists(id_, id_length, CKO_PUBLIC_KEY):
- raise DuplicationError("Public key with same ID already exists")
-
- # Process keyword boolean arguments
- (cka_copyable_ptr, cka_derive_ptr, cka_encrypt_ptr, cka_modifiable_ptr,
- cka_private_ptr, cka_trusted_ptr, cka_verify_ptr,
- cka_verify_recover_ptr, cka_wrap_ptr,) = convert_py2bool(attrs_pub)
-
- # decode from ASN1 DER
- try:
- pkey = serialization.load_der_public_key(data, default_backend())
- except Exception:
- raise Error("import_public_key: d2i_PUBKEY error")
- if isinstance(pkey, rsa.RSAPublicKey):
- ret = self._import_RSA_public_key(label, label_length, id_,
- id_length, pkey,
- cka_copyable_ptr,
- cka_derive_ptr,
- cka_encrypt_ptr,
- cka_modifiable_ptr,
- cka_private_ptr,
- cka_trusted_ptr,
- cka_verify_ptr,
- cka_verify_recover_ptr,
- cka_wrap_ptr)
- elif isinstance(pkey, dsa.DSAPublicKey):
- raise Error("DSA is not supported")
- elif isinstance(pkey, ec.EllipticCurvePublicKey):
- raise Error("EC is not supported")
- else:
- raise Error("Unsupported key type")
-
- return ret
-
- def export_wrapped_key(self, key, wrapping_key, wrapping_mech):
- """
- Export wrapped key
- """
- object_key = key
- object_wrapping_key = wrapping_key
- wrapped_key_len_ptr = new_ptr(CK_ULONG, 0)
- wrapping_mech_ptr = new_ptr(CK_MECHANISM, (wrapping_mech, NULL, 0))
- # currently we don't support parameter in mechanism
-
- # TODO check long overflow
- # TODO export method
-
- # fill mech parameters
- _set_wrapping_mech_parameters(wrapping_mech_ptr.mechanism,
- wrapping_mech_ptr)
-
- rv = self.p11.C_WrapKey(self.session, wrapping_mech_ptr,
- object_wrapping_key, object_key, NULL,
- wrapped_key_len_ptr)
- check_return_value(rv, "key wrapping: get buffer length")
-
- wrapped_key = new_array(CK_BYTE, wrapped_key_len_ptr[0])
-
- rv = self.p11.C_WrapKey(self.session, wrapping_mech_ptr,
- object_wrapping_key, object_key, wrapped_key,
- wrapped_key_len_ptr)
- check_return_value(rv, "key wrapping: wrapping")
-
- result = string_to_pybytes_or_none(wrapped_key, wrapped_key_len_ptr[0])
-
- return result
-
- def import_wrapped_secret_key(self, label, id, data, unwrapping_key,
- wrapping_mech, key_type, cka_copyable=True,
- cka_decrypt=False, cka_derive=False,
- cka_encrypt=False, cka_extractable=True,
- cka_modifiable=True, cka_private=True,
- cka_sensitive=True, cka_sign=False,
- cka_unwrap=True, cka_verify=False,
- cka_wrap=True, cka_wrap_with_trusted=False):
- """
- Import wrapped secret key
- """
- if isinstance(id, unicode):
- id = id.encode()
- if isinstance(data, unicode):
- data = data.encode()
-
- wrapped_key = new_array(CK_BYTE, data)
- wrapped_key_len = len(data)
- unwrapping_key_object = unwrapping_key
- unwrapped_key_object_ptr = new_ptr(CK_OBJECT_HANDLE, 0)
- label_unicode = label
- id_ = new_array(CK_BYTE, id)
- id_length = len(id)
- wrapping_mech_ptr = new_ptr(CK_MECHANISM, (wrapping_mech, NULL, 0))
- key_class_ptr = new_ptr(CK_OBJECT_CLASS, CKO_SECRET_KEY)
- key_type_ptr = new_ptr(CK_KEY_TYPE, key_type)
-
- attrs = (
- cka_copyable,
- cka_decrypt,
- cka_derive,
- cka_encrypt,
- cka_extractable,
- cka_modifiable,
- cka_private,
- cka_sensitive,
- cka_sign,
- cka_unwrap,
- cka_verify,
- cka_wrap,
- cka_wrap_with_trusted,
- )
-
- _set_wrapping_mech_parameters(wrapping_mech_ptr.mechanism,
- wrapping_mech_ptr)
-
- label, label_length = unicode_to_char_array(label_unicode)
-
- if self._id_exists(id_, id_length, key_class_ptr[0]):
- raise DuplicationError("Secret key with same ID already exists")
-
- # Process keyword boolean arguments
- (_cka_copyable_ptr, cka_decrypt_ptr, cka_derive_ptr, cka_encrypt_ptr,
- cka_extractable_ptr, cka_modifiable_ptr, cka_private_ptr,
- cka_sensitive_ptr, cka_sign_ptr, cka_unwrap_ptr, cka_verify_ptr,
- cka_wrap_ptr, cka_wrap_with_trusted_ptr,) = convert_py2bool(attrs)
-
- template = new_array(CK_ATTRIBUTE, (
- (CKA_CLASS, key_class_ptr, sizeof(CK_OBJECT_CLASS)),
- (CKA_KEY_TYPE, key_type_ptr, sizeof(CK_KEY_TYPE)),
- (CKA_ID, id_, id_length),
- (CKA_LABEL, label, label_length),
- (CKA_TOKEN, true_ptr, sizeof(CK_BBOOL)),
- # TODO Softhsm doesn't support it
- # (CKA_COPYABLE, cka_copyable_ptr, sizeof(CK_BBOOL)),
- (CKA_DECRYPT, cka_decrypt_ptr, sizeof(CK_BBOOL)),
- (CKA_DERIVE, cka_derive_ptr, sizeof(CK_BBOOL)),
- (CKA_ENCRYPT, cka_encrypt_ptr, sizeof(CK_BBOOL)),
- (CKA_EXTRACTABLE, cka_extractable_ptr, sizeof(CK_BBOOL)),
- (CKA_MODIFIABLE, cka_modifiable_ptr, sizeof(CK_BBOOL)),
- (CKA_PRIVATE, cka_private_ptr, sizeof(CK_BBOOL)),
- (CKA_SENSITIVE, cka_sensitive_ptr, sizeof(CK_BBOOL)),
- (CKA_SIGN, cka_sign_ptr, sizeof(CK_BBOOL)),
- (CKA_UNWRAP, cka_unwrap_ptr, sizeof(CK_BBOOL)),
- (CKA_VERIFY, cka_verify_ptr, sizeof(CK_BBOOL)),
- (CKA_WRAP, cka_wrap_ptr, sizeof(CK_BBOOL)),
- (CKA_WRAP_WITH_TRUSTED, cka_wrap_with_trusted_ptr,
- sizeof(CK_BBOOL)),
- ))
-
- rv = self.p11.C_UnwrapKey(self.session, wrapping_mech_ptr,
- unwrapping_key_object, wrapped_key,
- wrapped_key_len, template,
- sizeof(template) // sizeof(CK_ATTRIBUTE),
- unwrapped_key_object_ptr)
- check_return_value(rv, "import_wrapped_key: key unwrapping")
-
- return unwrapped_key_object_ptr[0]
-
- def import_wrapped_private_key(self, label, id, data, unwrapping_key,
- wrapping_mech, key_type,
- cka_always_authenticate=False,
- cka_copyable=True, cka_decrypt=False,
- cka_derive=False, cka_extractable=True,
- cka_modifiable=True, cka_private=True,
- cka_sensitive=True, cka_sign=True,
- cka_sign_recover=True, cka_unwrap=False,
- cka_wrap_with_trusted=False):
- """
- Import wrapped private key
- """
- if isinstance(id, unicode):
- id = id.encode()
- if isinstance(data, unicode):
- data = data.encode()
-
- wrapped_key = new_array(CK_BYTE, data)
- wrapped_key_len = len(data)
- unwrapping_key_object = unwrapping_key
- unwrapped_key_object_ptr = new_ptr(CK_OBJECT_HANDLE, 0)
- label_unicode = label
- id_ = new_array(CK_BYTE, id)
- id_length = len(id)
- wrapping_mech_ptr = new_ptr(CK_MECHANISM, (wrapping_mech, NULL, 0))
- key_class_ptr = new_ptr(CK_OBJECT_CLASS, CKO_PRIVATE_KEY)
- key_type_ptr = new_ptr(CK_KEY_TYPE, key_type)
-
- attrs_priv = (
- cka_always_authenticate,
- cka_copyable,
- cka_decrypt,
- cka_derive,
- cka_extractable,
- cka_modifiable,
- cka_private,
- cka_sensitive,
- cka_sign,
- cka_sign_recover,
- cka_unwrap,
- cka_wrap_with_trusted,
- )
-
- label, label_length = unicode_to_char_array(label_unicode)
-
- if self._id_exists(id_, id_length, CKO_SECRET_KEY):
- raise DuplicationError("Secret key with same ID already exists")
-
- # Process keyword boolean arguments
- (cka_always_authenticate_ptr, _cka_copyable_ptr, cka_decrypt_ptr,
- cka_derive_ptr, cka_extractable_ptr, cka_modifiable_ptr,
- cka_private_ptr, cka_sensitive_ptr, cka_sign_ptr,
- _cka_sign_recover_ptr, cka_unwrap_ptr, cka_wrap_with_trusted_ptr,
- ) = convert_py2bool(attrs_priv)
-
- template = new_array(CK_ATTRIBUTE, (
- (CKA_CLASS, key_class_ptr, sizeof(CK_OBJECT_CLASS)),
- (CKA_KEY_TYPE, key_type_ptr, sizeof(CK_KEY_TYPE)),
- (CKA_ID, id_, id_length),
- (CKA_LABEL, label, label_length),
- (CKA_TOKEN, true_ptr, sizeof(CK_BBOOL)),
- (CKA_ALWAYS_AUTHENTICATE, cka_always_authenticate_ptr,
- sizeof(CK_BBOOL)),
- # TODO Softhsm doesn't support it
- # (CKA_COPYABLE, cka_copyable_ptr, sizeof(CK_BBOOL)),
- (CKA_DECRYPT, cka_decrypt_ptr, sizeof(CK_BBOOL)),
- (CKA_DERIVE, cka_derive_ptr, sizeof(CK_BBOOL)),
- (CKA_EXTRACTABLE, cka_extractable_ptr, sizeof(CK_BBOOL)),
- (CKA_MODIFIABLE, cka_modifiable_ptr, sizeof(CK_BBOOL)),
- (CKA_PRIVATE, cka_private_ptr, sizeof(CK_BBOOL)),
- (CKA_SENSITIVE, cka_sensitive_ptr, sizeof(CK_BBOOL)),
- (CKA_SIGN, cka_sign_ptr, sizeof(CK_BBOOL)),
- (CKA_SIGN_RECOVER, cka_sign_ptr, sizeof(CK_BBOOL)),
- (CKA_UNWRAP, cka_unwrap_ptr, sizeof(CK_BBOOL)),
- (CKA_WRAP_WITH_TRUSTED, cka_wrap_with_trusted_ptr,
- sizeof(CK_BBOOL)),
- ))
-
- rv = self.p11.C_UnwrapKey(self.session, wrapping_mech_ptr,
- unwrapping_key_object, wrapped_key,
- wrapped_key_len, template,
- sizeof(template) // sizeof(CK_ATTRIBUTE),
- unwrapped_key_object_ptr)
- check_return_value(rv, "import_wrapped_key: key unwrapping")
-
- return unwrapped_key_object_ptr[0]
-
- def set_attribute(self, key_object, attr, value):
- """
- Set object attributes
- """
- object = key_object
- attribute_ptr = new_ptr(CK_ATTRIBUTE)
-
- attribute_ptr.type = attr
- if attr in (CKA_ALWAYS_AUTHENTICATE,
- CKA_ALWAYS_SENSITIVE,
- CKA_COPYABLE,
- CKA_ENCRYPT,
- CKA_EXTRACTABLE,
- CKA_DECRYPT,
- CKA_DERIVE,
- CKA_LOCAL,
- CKA_MODIFIABLE,
- CKA_NEVER_EXTRACTABLE,
- CKA_PRIVATE,
- CKA_SENSITIVE,
- CKA_SIGN,
- CKA_SIGN_RECOVER,
- CKA_TOKEN,
- CKA_TRUSTED,
- CKA_UNWRAP,
- CKA_VERIFY,
- CKA_VERIFY_RECOVER,
- CKA_WRAP,
- CKA_WRAP_WITH_TRUSTED):
- attribute_ptr.pValue = true_ptr if value else false_ptr
- attribute_ptr.ulValueLen = sizeof(CK_BBOOL)
- elif attr == CKA_ID:
- if not isinstance(value, bytes):
- raise Error("Bytestring value expected")
- attribute_ptr.pValue = new_array(CK_BYTE, value)
- attribute_ptr.ulValueLen = len(value)
- elif attr == CKA_LABEL:
- if not isinstance(value, unicode):
- raise Error("Unicode value expected")
- label, label_length = unicode_to_char_array(value)
- attribute_ptr.pValue = label
- attribute_ptr.ulValueLen = label_length
- elif attr == CKA_KEY_TYPE:
- if not isinstance(value, int):
- raise Error("Integer value expected")
- attribute_ptr.pValue = new_ptr(unsigned_long, value)
- attribute_ptr.ulValueLen = sizeof(unsigned_long)
- else:
- raise Error("Unknown attribute")
-
- template = new_array(CK_ATTRIBUTE, (attribute_ptr[0],))
-
- rv = self.p11.C_SetAttributeValue(self.session, object, template,
- (sizeof(template) //
- sizeof(CK_ATTRIBUTE)))
- check_return_value(rv, "set_attribute")
-
- def get_attribute(self, key_object, attr):
- object = key_object
- attribute_ptr = new_ptr(CK_ATTRIBUTE)
-
- attribute_ptr.type = attr
- attribute_ptr.pValue = NULL_PTR
- attribute_ptr.ulValueLen = 0
- template = new_array(CK_ATTRIBUTE, (attribute_ptr[0],))
-
- rv = self.p11.C_GetAttributeValue(self.session, object, template,
- (sizeof(template) //
- sizeof(CK_ATTRIBUTE)))
- if rv == CKR_ATTRIBUTE_TYPE_INVALID or template[0].ulValueLen == -1:
- raise NotFound("attribute does not exist")
- check_return_value(rv, "get_attribute init")
- value = new_array(unsigned_char, template[0].ulValueLen)
- template[0].pValue = value
-
- rv = self.p11.C_GetAttributeValue(self.session, object, template,
- (sizeof(template) //
- sizeof(CK_ATTRIBUTE)))
- check_return_value(rv, "get_attribute")
-
- if attr in (CKA_ALWAYS_AUTHENTICATE,
- CKA_ALWAYS_SENSITIVE,
- CKA_COPYABLE,
- CKA_ENCRYPT,
- CKA_EXTRACTABLE,
- CKA_DECRYPT,
- CKA_DERIVE,
- CKA_LOCAL,
- CKA_MODIFIABLE,
- CKA_NEVER_EXTRACTABLE,
- CKA_PRIVATE,
- CKA_SENSITIVE,
- CKA_SIGN,
- CKA_SIGN_RECOVER,
- CKA_TOKEN,
- CKA_TRUSTED,
- CKA_UNWRAP,
- CKA_VERIFY,
- CKA_VERIFY_RECOVER,
- CKA_WRAP,
- CKA_WRAP_WITH_TRUSTED):
- ret = bool(_ffi.cast(_ffi.getctype(CK_BBOOL, '*'), value)[0])
- elif attr == CKA_LABEL:
- ret = char_array_to_unicode(value, template[0].ulValueLen)
- elif attr in (CKA_MODULUS, CKA_PUBLIC_EXPONENT, CKA_ID):
- ret = string_to_pybytes_or_none(value, template[0].ulValueLen)
- elif attr == CKA_KEY_TYPE:
- ret = _ffi.cast(_ffi.getctype(unsigned_long, '*'), value)[0]
- else:
- raise Error("Unknown attribute")
-
- return ret
-
-
-# Key Classes
-KEY_CLASS_PUBLIC_KEY = CKO_PUBLIC_KEY
-KEY_CLASS_PRIVATE_KEY = CKO_PRIVATE_KEY
-KEY_CLASS_SECRET_KEY = CKO_SECRET_KEY
-
-# Key types
-KEY_TYPE_RSA = CKK_RSA
-KEY_TYPE_AES = CKK_AES
-
-# Wrapping mech type
-MECH_RSA_PKCS = CKM_RSA_PKCS
-MECH_RSA_PKCS_OAEP = CKM_RSA_PKCS_OAEP
-MECH_AES_KEY_WRAP = CKM_AES_KEY_WRAP
-MECH_AES_KEY_WRAP_PAD = CKM_AES_KEY_WRAP_PAD
-
-
-def generate_master_key(p11, keylabel=u"dnssec-master", key_length=16,
- disable_old_keys=True):
- assert isinstance(p11, P11_Helper)
-
- key_id = None
- while True:
- # check if key with this ID exist in LDAP or softHSM
- # id is 16 Bytes long
- key_id = "".join(chr(random.randint(0, 255)) for _ in range(0, 16))
- keys = p11.find_keys(KEY_CLASS_SECRET_KEY,
- label=keylabel,
- id=key_id)
- if not keys:
- break # we found unique id
-
- p11.generate_master_key(keylabel,
- key_id,
- key_length=key_length,
- cka_wrap=True,
- cka_unwrap=True)
-
- if disable_old_keys:
- # set CKA_WRAP=False for old master keys
- master_keys = p11.find_keys(KEY_CLASS_SECRET_KEY,
- label=keylabel,
- cka_wrap=True)
-
- for handle in master_keys:
- # don't disable wrapping for new key
- # compare IDs not handle
- if key_id != p11.get_attribute(handle, CKA_ID):
- p11.set_attribute(handle, CKA_WRAP, False)
diff --git a/ipapython/secrets/__init__.py b/ipapython/secrets/__init__.py
deleted file mode 100644
index e69de29bb..000000000
--- a/ipapython/secrets/__init__.py
+++ /dev/null
diff --git a/ipapython/secrets/client.py b/ipapython/secrets/client.py
deleted file mode 100644
index d9cc7d0f5..000000000
--- a/ipapython/secrets/client.py
+++ /dev/null
@@ -1,109 +0,0 @@
-# Copyright (C) 2015 IPA Project Contributors, see COPYING for license
-
-from __future__ import print_function
-from custodia.message.kem import KEMClient, KEY_USAGE_SIG, KEY_USAGE_ENC
-from jwcrypto.common import json_decode
-from jwcrypto.jwk import JWK
-from ipapython.secrets.kem import IPAKEMKeys
-from ipapython.secrets.store import iSecStore
-from ipaplatform.paths import paths
-from base64 import b64encode
-import ldapurl
-import gssapi
-import os
-import requests
-
-
-class CustodiaClient(object):
-
- def _client_keys(self):
- return self.ikk.server_keys
-
- def _server_keys(self, server, realm):
- principal = 'host/%s@%s' % (server, realm)
- sk = JWK(**json_decode(self.ikk.find_key(principal, KEY_USAGE_SIG)))
- ek = JWK(**json_decode(self.ikk.find_key(principal, KEY_USAGE_ENC)))
- return (sk, ek)
-
- def _ldap_uri(self, realm):
- dashrealm = '-'.join(realm.split('.'))
- socketpath = paths.SLAPD_INSTANCE_SOCKET_TEMPLATE % (dashrealm,)
- return 'ldapi://' + ldapurl.ldapUrlEscape(socketpath)
-
- def _keystore(self, realm, ldap_uri, auth_type):
- config = dict()
- if ldap_uri is None:
- config['ldap_uri'] = self._ldap_uri(realm)
- else:
- config['ldap_uri'] = ldap_uri
- if auth_type is not None:
- config['auth_type'] = auth_type
-
- return iSecStore(config)
-
- def __init__(
- self, client_service, keyfile, keytab, server, realm,
- ldap_uri=None, auth_type=None):
- self.client_service = client_service
- self.keytab = keytab
-
- # Init creds immediately to make sure they are valid. Creds
- # can also be re-inited by _auth_header to avoid expiry.
- #
- self.creds = self.init_creds()
-
- self.service_name = gssapi.Name('HTTP@%s' % (server,),
- gssapi.NameType.hostbased_service)
- self.server = server
-
- self.ikk = IPAKEMKeys({'server_keys': keyfile, 'ldap_uri': ldap_uri})
-
- self.kemcli = KEMClient(self._server_keys(server, realm),
- self._client_keys())
-
- self.keystore = self._keystore(realm, ldap_uri, auth_type)
-
- # FIXME: Remove warnings about missig subjAltName
- requests.packages.urllib3.disable_warnings()
-
- def init_creds(self):
- name = gssapi.Name(self.client_service,
- gssapi.NameType.hostbased_service)
- store = {'client_keytab': self.keytab,
- 'ccache': 'MEMORY:Custodia_%s' % b64encode(os.urandom(8))}
- return gssapi.Credentials(name=name, store=store, usage='initiate')
-
- def _auth_header(self):
- if not self.creds or self.creds.lifetime < 300:
- self.creds = self.init_creds()
- ctx = gssapi.SecurityContext(name=self.service_name, creds=self.creds)
- authtok = ctx.step()
- return {'Authorization': 'Negotiate %s' % b64encode(authtok)}
-
- def fetch_key(self, keyname, store=True):
-
- # Prepare URL
- url = 'https://%s/ipa/keys/%s' % (self.server, keyname)
-
- # Prepare signed/encrypted request
- encalg = ('RSA-OAEP', 'A256CBC-HS512')
- request = self.kemcli.make_request(keyname, encalg=encalg)
-
- # Prepare Authentication header
- headers = self._auth_header()
-
- # Perform request
- r = requests.get(url, headers=headers,
- params={'type': 'kem', 'value': request})
- r.raise_for_status()
- reply = r.json()
-
- if 'type' not in reply or reply['type'] != 'kem':
- raise RuntimeError('Invlid JSON response type')
-
- value = self.kemcli.parse_reply(keyname, reply['value'])
-
- if store:
- self.keystore.set('keys/%s' % keyname, value)
- else:
- return value
diff --git a/ipapython/secrets/common.py b/ipapython/secrets/common.py
deleted file mode 100644
index 2b906b649..000000000
--- a/ipapython/secrets/common.py
+++ /dev/null
@@ -1,45 +0,0 @@
-# Copyright (C) 2015 IPA Project Contributors, see COPYING for license
-from __future__ import print_function
-import ldap
-import ldap.sasl
-import ldap.filter
-
-
-class iSecLdap(object):
-
- def __init__(self, uri, auth_type=None):
- self.uri = uri
- if auth_type is not None:
- self.auth_type = auth_type
- else:
- if uri.startswith('ldapi'):
- self.auth_type = 'EXTERNAL'
- else:
- self.auth_type = 'GSSAPI'
- self._basedn = None
-
- @property
- def basedn(self):
- if self._basedn is None:
- conn = self.connect()
- r = conn.search_s('', ldap.SCOPE_BASE)
- self._basedn = r[0][1]['defaultnamingcontext'][0]
- return self._basedn
-
- def connect(self):
- conn = ldap.initialize(self.uri)
- if self.auth_type == 'EXTERNAL':
- auth_tokens = ldap.sasl.external(None)
- elif self.auth_type == 'GSSAPI':
- auth_tokens = ldap.sasl.sasl({}, 'GSSAPI')
- else:
- raise ValueError(
- 'Invalid authentication type: %s' % self.auth_type)
- conn.sasl_interactive_bind_s('', auth_tokens)
- return conn
-
- def build_filter(self, formatstr, args):
- escaped_args = dict()
- for key, value in args.iteritems():
- escaped_args[key] = ldap.filter.escape_filter_chars(value)
- return formatstr.format(**escaped_args)
diff --git a/ipapython/secrets/kem.py b/ipapython/secrets/kem.py
deleted file mode 100644
index 7f92c9f89..000000000
--- a/ipapython/secrets/kem.py
+++ /dev/null
@@ -1,228 +0,0 @@
-# Copyright (C) 2015 IPA Project Contributors, see COPYING for license
-
-from __future__ import print_function
-import os
-
-# pylint: disable=import-error
-from six.moves.configparser import ConfigParser
-# pylint: enable=import-error
-
-from ipaplatform.paths import paths
-from ipapython.dn import DN
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric import rsa, ec
-from custodia.message.kem import KEMKeysStore
-from custodia.message.kem import KEY_USAGE_SIG, KEY_USAGE_ENC, KEY_USAGE_MAP
-from jwcrypto.common import json_decode, json_encode
-from jwcrypto.common import base64url_encode
-from jwcrypto.jwk import JWK
-from ipapython.secrets.common import iSecLdap
-from binascii import unhexlify
-import ldap
-
-
-IPA_REL_BASE_DN = 'cn=custodia,cn=ipa,cn=etc'
-IPA_KEYS_QUERY = '(&(ipaKeyUsage={usage:s})(memberPrincipal={princ:s}))'
-RFC5280_USAGE_MAP = {KEY_USAGE_SIG: 'digitalSignature',
- KEY_USAGE_ENC: 'dataEncipherment'}
-
-
-class KEMLdap(iSecLdap):
-
- @property
- def keysbase(self):
- return '%s,%s' % (IPA_REL_BASE_DN, self.basedn)
-
- def _encode_int(self, i):
- I = hex(i).rstrip("L").lstrip("0x")
- return base64url_encode(unhexlify((len(I) % 2) * '0' + I))
-
- def _parse_public_key(self, ipa_public_key):
- public_key = serialization.load_der_public_key(ipa_public_key,
- default_backend())
- num = public_key.public_numbers()
- if isinstance(num, rsa.RSAPublicNumbers):
- return {'kty': 'RSA',
- 'e': self._encode_int(num.e),
- 'n': self._encode_int(num.n)}
- elif isinstance(num, ec.EllipticCurvePublicNumbers):
- if num.curve.name == 'secp256r1':
- curve = 'P-256'
- elif num.curve.name == 'secp384r1':
- curve = 'P-384'
- elif num.curve.name == 'secp521r1':
- curve = 'P-521'
- else:
- raise TypeError('Unsupported Elliptic Curve')
- return {'kty': 'EC',
- 'crv': curve,
- 'x': self._encode_int(num.x),
- 'y': self._encode_int(num.y)}
- else:
- raise TypeError('Unknown Public Key type')
-
- def get_key(self, usage, principal):
- conn = self.connect()
- scope = ldap.SCOPE_SUBTREE
-
- ldap_filter = self.build_filter(IPA_KEYS_QUERY,
- {'usage': RFC5280_USAGE_MAP[usage],
- 'princ': principal})
- r = conn.search_s(self.keysbase, scope, ldap_filter)
- if len(r) != 1:
- raise ValueError("Incorrect number of results (%d) searching for"
- "public key for %s" % (len(r), principal))
- ipa_public_key = r[0][1]['ipaPublicKey'][0]
- jwk = self._parse_public_key(ipa_public_key)
- jwk['use'] = KEY_USAGE_MAP[usage]
- return json_encode(jwk)
-
- def _format_public_key(self, key):
- if isinstance(key, str):
- jwkey = json_decode(key)
- if 'kty' not in jwkey:
- raise ValueError('Invalid key, missing "kty" attribute')
- if jwkey['kty'] == 'RSA':
- pubnum = rsa.RSAPublicNumbers(jwkey['e'], jwkey['n'])
- pubkey = pubnum.public_key(default_backend())
- elif jwkey['kty'] == 'EC':
- if jwkey['crv'] == 'P-256':
- curve = ec.SECP256R1
- elif jwkey['crv'] == 'P-384':
- curve = ec.SECP384R1
- elif jwkey['crv'] == 'P-521':
- curve = ec.SECP521R1
- else:
- raise TypeError('Unsupported Elliptic Curve')
- pubnum = ec.EllipticCurvePublicNumbers(
- jwkey['x'], jwkey['y'], curve)
- pubkey = pubnum.public_key(default_backend())
- else:
- raise ValueError('Unknown key type: %s' % jwkey['kty'])
- elif isinstance(key, rsa.RSAPublicKey):
- pubkey = key
- elif isinstance(key, ec.EllipticCurvePublicKey):
- pubkey = key
- else:
- raise TypeError('Unknown key type: %s' % type(key))
-
- return pubkey.public_bytes(
- encoding=serialization.Encoding.DER,
- format=serialization.PublicFormat.SubjectPublicKeyInfo)
-
- def set_key(self, usage, principal, key):
- """
- Write key for the host or service.
-
- Service keys are nested one level beneath the 'cn=custodia'
- container, in the 'cn=<servicename>' container; this allows
- fine-grained control over key management permissions for
- specific services.
-
- The container is assumed to exist.
-
- """
- public_key = self._format_public_key(key)
- conn = self.connect()
- servicename, host = principal.split('@')[0].split('/')
- name = '%s/%s' % (KEY_USAGE_MAP[usage], host)
- service_rdn = ('cn', servicename) if servicename != 'host' else DN()
- dn = str(DN(('cn', name), service_rdn, self.keysbase))
- try:
- mods = [('objectClass', ['nsContainer',
- 'ipaKeyPolicy',
- 'ipaPublicKeyObject',
- 'groupOfPrincipals']),
- ('cn', name),
- ('ipaKeyUsage', RFC5280_USAGE_MAP[usage]),
- ('memberPrincipal', principal),
- ('ipaPublicKey', public_key)]
- conn.add_s(dn, mods)
- except Exception: # pylint: disable=broad-except
- # This may fail if the entry already exists
- mods = [(ldap.MOD_REPLACE, 'ipaPublicKey', public_key)]
- conn.modify_s(dn, mods)
-
-
-def newServerKeys(path, keyid):
- skey = JWK(generate='RSA', use='sig', kid=keyid)
- ekey = JWK(generate='RSA', use='enc', kid=keyid)
- with open(path, 'w') as f:
- os.fchmod(f.fileno(), 0o600)
- os.fchown(f.fileno(), 0, 0)
- f.write('[%s,%s]' % (skey.export(), ekey.export()))
- return [skey.get_op_key('verify'), ekey.get_op_key('encrypt')]
-
-
-class IPAKEMKeys(KEMKeysStore):
- """A KEM Keys Store.
-
- This is a store that holds public keys of registered
- clients allowed to use KEM messages. It takes the form
- of an authorizer merely for the purpose of attaching
- itself to a 'request' so that later on the KEM Parser
- can fetch the appropariate key to verify/decrypt an
- incoming request and make the payload available.
-
- The KEM Parser will actually perform additional
- authorization checks in this case.
-
- SimplePathAuthz is extended here as we want to attach the
- store only to requests on paths we are configured to
- manage.
- """
-
- def __init__(self, config=None, ipaconf=paths.IPA_DEFAULT_CONF):
- super(IPAKEMKeys, self).__init__(config)
- conf = ConfigParser()
- conf.read(ipaconf)
- self.host = conf.get('global', 'host')
- self.realm = conf.get('global', 'realm')
- self.ldap_uri = config.get('ldap_uri', None)
- if self.ldap_uri is None:
- self.ldap_uri = conf.get('global', 'ldap_uri', None)
- self._server_keys = None
-
- def find_key(self, kid, usage):
- if kid is None:
- raise TypeError('Key ID is None, should be a SPN')
- conn = KEMLdap(self.ldap_uri)
- return conn.get_key(usage, kid)
-
- def generate_server_keys(self):
- self.generate_keys('host')
-
- def generate_keys(self, servicename):
- principal = '%s/%s@%s' % (servicename, self.host, self.realm)
- # Neutralize the key with read if any
- self._server_keys = None
- # Generate private key and store it
- pubkeys = newServerKeys(self.config['server_keys'], principal)
- # Store public key in LDAP
- ldapconn = KEMLdap(self.ldap_uri)
- ldapconn.set_key(KEY_USAGE_SIG, principal, pubkeys[0])
- ldapconn.set_key(KEY_USAGE_ENC, principal, pubkeys[1])
-
- @property
- def server_keys(self):
- if self._server_keys is None:
- with open(self.config['server_keys']) as f:
- jsonkeys = f.read()
- dictkeys = json_decode(jsonkeys)
- self._server_keys = (JWK(**dictkeys[KEY_USAGE_SIG]),
- JWK(**dictkeys[KEY_USAGE_ENC]))
- return self._server_keys
-
-
-# Manual testing
-if __name__ == '__main__':
- IKK = IPAKEMKeys({'paths': '/',
- 'server_keys': '/etc/ipa/custodia/server.keys'})
- IKK.generate_server_keys()
- print(('SIG', IKK.server_keys[0].export_public()))
- print(('ENC', IKK.server_keys[1].export_public()))
- print(IKK.find_key('host/%s@%s' % (IKK.host, IKK.realm),
- usage=KEY_USAGE_SIG))
- print(IKK.find_key('host/%s@%s' % (IKK.host, IKK.realm),
- usage=KEY_USAGE_ENC))
diff --git a/ipapython/secrets/store.py b/ipapython/secrets/store.py
deleted file mode 100644
index 30a87d4a5..000000000
--- a/ipapython/secrets/store.py
+++ /dev/null
@@ -1,261 +0,0 @@
-# Copyright (C) 2015 IPA Project Contributors, see COPYING for license
-
-from __future__ import print_function
-from base64 import b64encode, b64decode
-from custodia.store.interface import CSStore
-from jwcrypto.common import json_decode, json_encode
-from ipaplatform.paths import paths
-from ipapython import ipautil
-from ipapython.secrets.common import iSecLdap
-import ldap
-import os
-import shutil
-import sys
-import tempfile
-
-
-class UnknownKeyName(Exception):
- pass
-
-
-class DBMAPHandler(object):
-
- def __init__(self, config, dbmap, nickname):
- raise NotImplementedError
-
- def export_key(self):
- raise NotImplementedError
-
- def import_key(self, value):
- raise NotImplementedError
-
-
-def log_error(error):
- print(error, file=sys.stderr)
-
-
-def PKI_TOMCAT_password_callback():
- password = None
- with open(paths.PKI_TOMCAT_PASSWORD_CONF) as f:
- for line in f.readlines():
- key, value = line.strip().split('=')
- if key == 'internal':
- password = value
- break
- return password
-
-
-def HTTPD_password_callback():
- with open(paths.ALIAS_PWDFILE_TXT) as f:
- password = f.read()
- return password
-
-
-class NSSWrappedCertDB(DBMAPHandler):
- '''
- Store that extracts private keys from an NSSDB, wrapped with the
- private key of the primary CA.
- '''
-
- def __init__(self, config, dbmap, nickname):
- if 'path' not in dbmap:
- raise ValueError(
- 'Configuration does not provide NSSDB path')
- if 'pwcallback' not in dbmap:
- raise ValueError(
- 'Configuration does not provide Password Calback')
- if 'wrap_nick' not in dbmap:
- raise ValueError(
- 'Configuration does not provide nickname of wrapping key')
- self.nssdb_path = dbmap['path']
- self.nssdb_password = dbmap['pwcallback']()
- self.wrap_nick = dbmap['wrap_nick']
- self.target_nick = nickname
-
- def export_key(self):
- tdir = tempfile.mkdtemp(dir=paths.TMP)
- try:
- nsspwfile = os.path.join(tdir, 'nsspwfile')
- with open(nsspwfile, 'w+') as f:
- f.write(self.nssdb_password)
- wrapped_key_file = os.path.join(tdir, 'wrapped_key')
- certificate_file = os.path.join(tdir, 'certificate')
- ipautil.run([
- paths.PKI, '-d', self.nssdb_path, '-C', nsspwfile,
- 'ca-authority-key-export',
- '--wrap-nickname', self.wrap_nick,
- '--target-nickname', self.target_nick,
- '-o', wrapped_key_file])
- ipautil.run([
- paths.CERTUTIL, '-d', self.nssdb_path,
- '-L', '-n', self.target_nick,
- '-a', '-o', certificate_file])
- with open(wrapped_key_file, 'r') as f:
- wrapped_key = f.read()
- with open(certificate_file, 'r') as f:
- certificate = f.read()
- finally:
- shutil.rmtree(tdir)
- return json_encode({
- 'wrapped_key': b64encode(wrapped_key),
- 'certificate': certificate})
-
-
-class NSSCertDB(DBMAPHandler):
-
- def __init__(self, config, dbmap, nickname):
- if 'type' not in dbmap or dbmap['type'] != 'NSSDB':
- raise ValueError('Invalid type "%s",'
- ' expected "NSSDB"' % (dbmap['type'],))
- if 'path' not in dbmap:
- raise ValueError('Configuration does not provide NSSDB path')
- if 'pwcallback' not in dbmap:
- raise ValueError('Configuration does not provide Password Calback')
- self.nssdb_path = dbmap['path']
- self.nickname = nickname
- self.nssdb_password = dbmap['pwcallback']()
-
- def export_key(self):
- tdir = tempfile.mkdtemp(dir=paths.TMP)
- try:
- nsspwfile = os.path.join(tdir, 'nsspwfile')
- with open(nsspwfile, 'w+') as f:
- f.write(self.nssdb_password)
- pk12pwfile = os.path.join(tdir, 'pk12pwfile')
- password = b64encode(os.urandom(16))
- with open(pk12pwfile, 'w+') as f:
- f.write(password)
- pk12file = os.path.join(tdir, 'pk12file')
- ipautil.run([paths.PK12UTIL,
- "-d", self.nssdb_path,
- "-o", pk12file,
- "-n", self.nickname,
- "-k", nsspwfile,
- "-w", pk12pwfile])
- with open(pk12file, 'r') as f:
- data = f.read()
- finally:
- shutil.rmtree(tdir)
- return json_encode({'export password': password,
- 'pkcs12 data': b64encode(data)})
-
- def import_key(self, value):
- v = json_decode(value)
- tdir = tempfile.mkdtemp(dir=paths.TMP)
- try:
- nsspwfile = os.path.join(tdir, 'nsspwfile')
- with open(nsspwfile, 'w+') as f:
- f.write(self.nssdb_password)
- pk12pwfile = os.path.join(tdir, 'pk12pwfile')
- with open(pk12pwfile, 'w+') as f:
- f.write(v['export password'])
- pk12file = os.path.join(tdir, 'pk12file')
- with open(pk12file, 'w+') as f:
- f.write(b64decode(v['pkcs12 data']))
- ipautil.run([paths.PK12UTIL,
- "-d", self.nssdb_path,
- "-i", pk12file,
- "-n", self.nickname,
- "-k", nsspwfile,
- "-w", pk12pwfile])
- finally:
- shutil.rmtree(tdir)
-
-
-# Exfiltrate the DM password Hash so it can be set in replica's and this
-# way let a replica be install without knowing the DM password and yet
-# still keep the DM password synchronized across replicas
-class DMLDAP(DBMAPHandler):
-
- def __init__(self, config, dbmap, nickname):
- if 'type' not in dbmap or dbmap['type'] != 'DMLDAP':
- raise ValueError('Invalid type "%s",'
- ' expected "DMLDAP"' % (dbmap['type'],))
- if nickname != 'DMHash':
- raise UnknownKeyName("Unknown Key Named '%s'" % nickname)
- self.ldap = iSecLdap(config['ldap_uri'],
- config.get('auth_type', None))
-
- def export_key(self):
- conn = self.ldap.connect()
- r = conn.search_s('cn=config', ldap.SCOPE_BASE,
- attrlist=['nsslapd-rootpw'])
- if len(r) != 1:
- raise RuntimeError('DM Hash not found!')
- return json_encode({'dmhash': r[0][1]['nsslapd-rootpw'][0]})
-
- def import_key(self, value):
- v = json_decode(value)
- conn = self.ldap.connect()
- mods = [(ldap.MOD_REPLACE, 'nsslapd-rootpw', str(v['dmhash']))]
- conn.modify_s('cn=config', mods)
-
-
-NAME_DB_MAP = {
- 'ca': {
- 'type': 'NSSDB',
- 'path': paths.PKI_TOMCAT_ALIAS_DIR,
- 'handler': NSSCertDB,
- 'pwcallback': PKI_TOMCAT_password_callback,
- },
- 'ca_wrapped': {
- 'handler': NSSWrappedCertDB,
- 'path': paths.PKI_TOMCAT_ALIAS_DIR,
- 'pwcallback': PKI_TOMCAT_password_callback,
- 'wrap_nick': 'caSigningCert cert-pki-ca',
- },
- 'ra': {
- 'type': 'NSSDB',
- 'path': paths.HTTPD_ALIAS_DIR,
- 'handler': NSSCertDB,
- 'pwcallback': HTTPD_password_callback,
- },
- 'dm': {
- 'type': 'DMLDAP',
- 'handler': DMLDAP,
- }
-}
-
-
-class IPASecStore(CSStore):
-
- def __init__(self, config=None):
- self.config = config
-
- def _get_handler(self, key):
- path = key.split('/', 3)
- if len(path) != 3 or path[0] != 'keys':
- raise ValueError('Invalid name')
- if path[1] not in NAME_DB_MAP:
- raise UnknownKeyName("Unknown DB named '%s'" % path[1])
- dbmap = NAME_DB_MAP[path[1]]
- return dbmap['handler'](self.config, dbmap, path[2])
-
- def get(self, key):
- try:
- key_handler = self._get_handler(key)
- value = key_handler.export_key()
- except Exception as e: # pylint: disable=broad-except
- log_error('Error retrievieng key "%s": %s' % (key, str(e)))
- value = None
- return value
-
- def set(self, key, value, replace=False):
- try:
- key_handler = self._get_handler(key)
- key_handler.import_key(value)
- except Exception as e: # pylint: disable=broad-except
- log_error('Error storing key "%s": %s' % (key, str(e)))
-
- def list(self, keyfilter=None):
- raise NotImplementedError
-
- def cut(self, key):
- raise NotImplementedError
-
- def span(self, key):
- raise NotImplementedError
-
-
-# backwards compatibility with FreeIPA 4.3 and 4.4.
-iSecStore = IPASecStore
diff --git a/ipapython/setup.py b/ipapython/setup.py
index 087086eee..1abe7b067 100755
--- a/ipapython/setup.py
+++ b/ipapython/setup.py
@@ -34,8 +34,6 @@ if __name__ == '__main__':
package_dir={'ipapython': ''},
packages=[
"ipapython",
- "ipapython.dnssec",
- "ipapython.secrets",
"ipapython.install"
],
install_requires=[
@@ -60,12 +58,4 @@ if __name__ == '__main__':
extras_require={
":python_version<'3'": ["enum34"],
},
- entry_points={
- 'custodia.authorizers': [
- 'IPAKEMKeys = ipapython.secrets.kem:IPAKEMKeys',
- ],
- 'custodia.stores': [
- 'IPASecStore = ipapython.secrets.store:IPASecStore',
- ],
- },
)
generated by cgit (git 2.34.1) at 2024-11-16 11:17:48 +0000