IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Migrate OTP import script to python-cryptography

Browse Source 
https://fedorahosted.org/freeipa/ticket/5192

Reviewed-By: Stanislav Laznicka <slaznick@redhat.com>
This commit is contained in:
Nathaniel McCallum 2015-08-31 10:46:19 -04:00 committed by Martin Basti
parent 3be696c92f
commit d00ae870dd
2 changed files with 80 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
ipaserver/install/ipa_otptoken_import.py
View File

@ -29,11 +29,15 @@ import struct
from lxml import etree from lxml import etree
import dateutil.parser import dateutil.parser
import dateutil.tz import dateutil.tz
import nss.nss as nss
import gssapi import gssapi
import six import six
from six.moves import xrange from six.moves import xrange
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf import pbkdf2
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from ipaplatform.paths import paths from ipaplatform.paths import paths
from ipapython import admintool from ipapython import admintool
from ipalib import api, errors from ipalib import api, errors
@ -119,13 +123,13 @@ def convertAlgorithm(value):
"Converts encryption URI to (mech, ivlen)." "Converts encryption URI to (mech, ivlen)."
return { return {
"http://www.w3.org/2001/04/xmlenc#aes128-cbc": (nss.CKM_AES_CBC_PAD, 128), "http://www.w3.org/2001/04/xmlenc#aes128-cbc": (algorithms.AES, modes.CBC, 128),
"http://www.w3.org/2001/04/xmlenc#aes192-cbc": (nss.CKM_AES_CBC_PAD, 192), "http://www.w3.org/2001/04/xmlenc#aes192-cbc": (algorithms.AES, modes.CBC, 192),
"http://www.w3.org/2001/04/xmlenc#aes256-cbc": (nss.CKM_AES_CBC_PAD, 256), "http://www.w3.org/2001/04/xmlenc#aes256-cbc": (algorithms.AES, modes.CBC, 256),
"http://www.w3.org/2001/04/xmlenc#tripledes-cbc": (nss.CKM_DES3_CBC_PAD, 64), "http://www.w3.org/2001/04/xmlenc#tripledes-cbc": (algorithms.TripleDES, modes.CBC, 64),
"http://www.w3.org/2001/04/xmldsig-more#camellia128": (nss.CKM_CAMELLIA_CBC_PAD, 128), "http://www.w3.org/2001/04/xmldsig-more#camellia128": (algorithms.Camellia, modes.CBC, 128),
"http://www.w3.org/2001/04/xmldsig-more#camellia192": (nss.CKM_CAMELLIA_CBC_PAD, 192), "http://www.w3.org/2001/04/xmldsig-more#camellia192": (algorithms.Camellia, modes.CBC, 192),
"http://www.w3.org/2001/04/xmldsig-more#camellia256": (nss.CKM_CAMELLIA_CBC_PAD, 256), "http://www.w3.org/2001/04/xmldsig-more#camellia256": (algorithms.Camellia, modes.CBC, 256),
# TODO: add support for these formats. # TODO: add support for these formats.
# "http://www.w3.org/2001/04/xmlenc#kw-aes128": "kw-aes128", # "http://www.w3.org/2001/04/xmlenc#kw-aes128": "kw-aes128",
@ -135,7 +139,7 @@ def convertAlgorithm(value):
# "http://www.w3.org/2001/04/xmldsig-more#kw-camellia128": "kw-camellia128", # "http://www.w3.org/2001/04/xmldsig-more#kw-camellia128": "kw-camellia128",
# "http://www.w3.org/2001/04/xmldsig-more#kw-camellia192": "kw-camellia192", # "http://www.w3.org/2001/04/xmldsig-more#kw-camellia192": "kw-camellia192",
# "http://www.w3.org/2001/04/xmldsig-more#kw-camellia256": "kw-camellia256", # "http://www.w3.org/2001/04/xmldsig-more#kw-camellia256": "kw-camellia256",
}.get(value.lower(), (None, None)) }.get(value.lower(), (None, None, None))
def convertEncrypted(value, decryptor=None, pconv=base64.b64decode, econv=lambda x: x): def convertEncrypted(value, decryptor=None, pconv=base64.b64decode, econv=lambda x: x):
@ -170,50 +174,29 @@ class PBKDF2KeyDerivation(XMLKeyDerivation):
if params is None: if params is None:
raise ValueError("XML file is missing PBKDF2 parameters!") raise ValueError("XML file is missing PBKDF2 parameters!")
self.salt = fetch(params, "./xenc11:Salt/xenc11:Specified/text()", base64.b64decode) salt = fetch(params, "./xenc11:Salt/xenc11:Specified/text()", base64.b64decode)
self.iter = fetch(params, "./xenc11:IterationCount/text()", int) itrs = fetch(params, "./xenc11:IterationCount/text()", int)
self.klen = fetch(params, "./xenc11:KeyLength/text()", int) klen = fetch(params, "./xenc11:KeyLength/text()", int)
self.hmod = fetch(params, "./xenc11:PRF/@Algorithm", convertHMACType, hashlib.sha1) hmod = fetch(params, "./xenc11:PRF/@Algorithm", convertHMACType, hashlib.sha1)
if self.salt is None: if salt is None:
raise ValueError("XML file is missing PBKDF2 salt!") raise ValueError("XML file is missing PBKDF2 salt!")
if self.iter is None: if itrs is None:
raise ValueError("XML file is missing PBKDF2 iteration count!") raise ValueError("XML file is missing PBKDF2 iteration count!")
if self.klen is None: if klen is None:
raise ValueError("XML file is missing PBKDF2 key length!") raise ValueError("XML file is missing PBKDF2 key length!")
self.kdf = pbkdf2.PBKDF2HMAC(
algorithm=hmod,
length=klen,
salt=salt,
iterations=itrs
)
def derive(self, masterkey): def derive(self, masterkey):
mac = hmac.HMAC(masterkey, None, self.hmod) return self.kdf.derive(masterkey)
# Figure out how many blocks we will have to combine
# to expand the master key to the desired length.
blocks = self.klen // mac.digest_size
if self.klen % mac.digest_size != 0:
blocks += 1
# Loop through each block adding it to the derived key.
dk = []
for i in range(1, blocks + 1):
# Set initial values.
last = self.salt + struct.pack('>I', i)
hash = [0] * mac.digest_size
# Perform n iterations.
for _j in xrange(self.iter):
tmp = mac.copy()
tmp.update(last)
last = tmp.digest()
# XOR the previous hash with the new hash.
for k in range(mac.digest_size):
hash[k] ^= ord(last[k])
# Add block to derived key.
dk.extend(hash)
return ''.join([chr(c) for c in dk])[:self.klen]
def convertKeyDerivation(value): def convertKeyDerivation(value):
@ -230,13 +213,17 @@ class XMLDecryptor(object):
* RFC 6931""" * RFC 6931"""
def __init__(self, key, hmac=None): def __init__(self, key, hmac=None):
self.__key = nss.SecItem(key) self.__key = key
self.__hmac = hmac self.__hmac = hmac
def __call__(self, element, mac=None): def __call__(self, element, mac=None):
(mech, ivlen) = fetch(element, "./xenc:EncryptionMethod/@Algorithm", convertAlgorithm) (algo, mode, klen) = fetch(element, "./xenc:EncryptionMethod/@Algorithm", convertAlgorithm)
data = fetch(element, "./xenc:CipherData/xenc:CipherValue/text()", base64.b64decode) data = fetch(element, "./xenc:CipherData/xenc:CipherValue/text()", base64.b64decode)
# Make sure the key is the right length.
if len(self.__key) * 8 != klen:
raise ValidationError("Invalid key length!")
# If a MAC is present, perform validation. # If a MAC is present, perform validation.
if mac: if mac:
tmp = self.__hmac.copy() tmp = self.__hmac.copy()
@ -244,13 +231,14 @@ class XMLDecryptor(object):
if tmp.digest() != mac: if tmp.digest() != mac:
raise ValidationError("MAC validation failed!") raise ValidationError("MAC validation failed!")
# Decrypt the data. iv = data[:algo.block_size / 8]
slot = nss.get_best_slot(mech) data = data[len(iv):]
key = nss.import_sym_key(slot, mech, nss.PK11_OriginUnwrap, nss.CKA_ENCRYPT, self.__key)
iv = nss.param_from_iv(mech, nss.SecItem(data[0:ivlen//8])) cipher = Cipher(algo(self.__key), mode(iv))
ctx = nss.create_context_by_sym_key(mech, nss.CKA_DECRYPT, key, iv) decryptor = cipher.decryptor()
out = ctx.cipher_op(data[ivlen // 8:])
out += ctx.digest_final() out = decryptor.update(data)
out += decryptor.finalize()
return out return out
@ -469,14 +457,6 @@ class OTPTokenImport(admintool.AdminTool):
description = "Import OTP tokens." description = "Import OTP tokens."
usage = "%prog [options] <PSKC file> <output file>" usage = "%prog [options] <PSKC file> <output file>"
@classmethod
def main(cls, argv):
nss.nss_init_nodb()
try:
super(OTPTokenImport, cls).main(argv)
finally:
nss.nss_shutdown()
@classmethod @classmethod
def add_options(cls, parser): def add_options(cls, parser):
super(OTPTokenImport, cls).add_options(parser) super(OTPTokenImport, cls).add_options(parser)

100
ipatests/test_ipaserver/test_otptoken_import.py
View File

@ -19,16 +19,13 @@
import os import os
import pytest import pytest
from nss import nss
from ipaserver.install.ipa_otptoken_import import PSKCDocument, ValidationError from ipaserver.install.ipa_otptoken_import import PSKCDocument, ValidationError
basename = os.path.join(os.path.dirname(__file__), "data") basename = os.path.join(os.path.dirname(__file__), "data")
@pytest.mark.skipif(True, reason="Causes NSS errors. Ticket 5192")
@pytest.mark.tier1 @pytest.mark.tier1
class test_otptoken_import(object): class test_otptoken_import(object):
def test_figure3(self): def test_figure3(self):
doc = PSKCDocument(os.path.join(basename, "pskc-figure3.xml")) doc = PSKCDocument(os.path.join(basename, "pskc-figure3.xml"))
assert doc.keyname is None assert doc.keyname is None
@ -63,62 +60,47 @@ class test_otptoken_import(object):
assert False assert False
def test_figure6(self): def test_figure6(self):
nss.nss_init_nodb() doc = PSKCDocument(os.path.join(basename, "pskc-figure6.xml"))
try: assert doc.keyname == 'Pre-shared-key'
doc = PSKCDocument(os.path.join(basename, "pskc-figure6.xml")) doc.setKey('12345678901234567890123456789012'.decode('hex'))
assert doc.keyname == 'Pre-shared-key' assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
doc.setKey('12345678901234567890123456789012'.decode('hex')) [(u'12345678', {
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \ 'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
[(u'12345678', { 'ipatokenvendor': u'Manufacturer',
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ', 'ipatokenserial': u'987654321',
'ipatokenvendor': u'Manufacturer', 'ipatokenhotpcounter': 0,
'ipatokenserial': u'987654321', 'ipatokenotpdigits': 8,
'ipatokenhotpcounter': 0, 'type': u'hotp'})]
'ipatokenotpdigits': 8,
'type': u'hotp'})]
finally:
nss.nss_shutdown()
def test_figure7(self): def test_figure7(self):
nss.nss_init_nodb() doc = PSKCDocument(os.path.join(basename, "pskc-figure7.xml"))
try: assert doc.keyname == 'My Password 1'
doc = PSKCDocument(os.path.join(basename, "pskc-figure7.xml")) doc.setKey('qwerty')
assert doc.keyname == 'My Password 1' assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
doc.setKey('qwerty') [(u'123456', {
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \ 'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
[(u'123456', { 'ipatokenvendor': u'TokenVendorAcme',
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ', 'ipatokenserial': u'987654321',
'ipatokenvendor': u'TokenVendorAcme', 'ipatokenotpdigits': 8,
'ipatokenserial': u'987654321', 'type': u'hotp'})]
'ipatokenotpdigits': 8,
'type': u'hotp'})]
finally:
nss.nss_shutdown()
def test_figure8(self): def test_figure8(self):
nss.nss_init_nodb()
try: try:
PSKCDocument(os.path.join(basename, "pskc-figure8.xml")) PSKCDocument(os.path.join(basename, "pskc-figure8.xml"))
except NotImplementedError: # X.509 is not supported. except NotImplementedError: # X.509 is not supported.
pass pass
else: else:
assert False assert False
finally:
nss.nss_shutdown()
def test_invalid(self): def test_invalid(self):
nss.nss_init_nodb()
try: try:
PSKCDocument(os.path.join(basename, "pskc-invalid.xml")) PSKCDocument(os.path.join(basename, "pskc-invalid.xml"))
except ValueError: # File is invalid. except ValueError: # File is invalid.
pass pass
else: else:
assert False assert False
finally:
nss.nss_shutdown()
def test_mini(self): def test_mini(self):
nss.nss_init_nodb()
try: try:
doc = PSKCDocument(os.path.join(basename, "pskc-mini.xml")) doc = PSKCDocument(os.path.join(basename, "pskc-mini.xml"))
for t in doc.getKeyPackages(): for t in doc.getKeyPackages():
@ -127,28 +109,22 @@ class test_otptoken_import(object):
pass pass
else: else:
assert False assert False
finally:
nss.nss_shutdown()
def test_full(self): def test_full(self):
nss.nss_init_nodb() doc = PSKCDocument(os.path.join(basename, "full.xml"))
try: assert [(t.id, t.options) for t in doc.getKeyPackages()] == \
doc = PSKCDocument(os.path.join(basename, "full.xml")) [(u'KID1', {
assert [(t.id, t.options) for t in doc.getKeyPackages()] == \ 'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ',
[(u'KID1', { 'ipatokennotafter': u'20060531000000Z',
'ipatokenotpkey': u'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ', 'ipatokennotbefore': u'20060501000000Z',
'ipatokennotafter': u'20060531000000Z', 'ipatokenserial': u'SerialNo-IssueNo',
'ipatokennotbefore': u'20060501000000Z', 'ipatokentotpclockoffset': 60000,
'ipatokenserial': u'SerialNo-IssueNo', 'ipatokenotpalgorithm': u'sha1',
'ipatokentotpclockoffset': 60000, 'ipatokenvendor': u'iana.dummy',
'ipatokenotpalgorithm': u'sha1', 'description': u'FriendlyName',
'ipatokenvendor': u'iana.dummy', 'ipatokentotptimestep': 200,
'description': u'FriendlyName', 'ipatokenhotpcounter': 0,
'ipatokentotptimestep': 200, 'ipatokenmodel': u'Model',
'ipatokenhotpcounter': 0, 'ipatokenotpdigits': 8,
'ipatokenmodel': u'Model', 'type': u'hotp',
'ipatokenotpdigits': 8, })]
'type': u'hotp',
})]
finally:
nss.nss_shutdown()