From ffeb5ef066942c2163aed7f1332a047d8c382f82 Mon Sep 17 00:00:00 2001 From: oepen Date: Wed, 17 Aug 2011 10:31:18 +0000 Subject: [PATCH] Refactoring: Merged Secure_Messaging and Security_Environment classes git-svn-id: https://vsmartcard.svn.sourceforge.net/svnroot/vsmartcard@489 96b47cad-a561-4643-ad3b-153ac7d7599c --- .../vpicc/virtualsmartcard/SmartcardSAM.py | 248 ++++++++---------- 1 file changed, 116 insertions(+), 132 deletions(-) diff --git a/virtualsmartcard/src/vpicc/virtualsmartcard/SmartcardSAM.py b/virtualsmartcard/src/vpicc/virtualsmartcard/SmartcardSAM.py index 2a7d37b..e68f80e 100644 --- a/virtualsmartcard/src/vpicc/virtualsmartcard/SmartcardSAM.py +++ b/virtualsmartcard/src/vpicc/virtualsmartcard/SmartcardSAM.py @@ -82,11 +82,13 @@ class SAM(object): else: self.cardSecret = cardSecret - self.SM_handler = Secure_Messaging(self.mf, self) + #Security Environments may be saved and retrieved from/to this dictionary + self.saved_SEs = {} + self.current_SE = Security_Environment(self.mf, self) def set_MF(self, mf): self.mf = mf - self.SM_handler.set_MF(mf) + self.current_SE.set_MF(mf) def FSencrypt(self, data): """ @@ -104,6 +106,45 @@ class SAM(object): """ return data + def store_SE(self, SEID): + """ + Stores the current Security environment in the secure access module. The + SEID is used as a reference to identify the SE. + """ + SEstr = dumps(self.current_SE) + self.saved_SEs[SEID] = SEstr + return SW["NORMAL"], "" + + def restore_SE(self, SEID): + """ + Restores a Security Environment from the SAM and replaces the current SE + with it + """ + + if (not self.saved_SEs.has_key(SEID)): + raise SwError(SW["ERR_REFNOTUSABLE"]) + else: + SEstr = self.saved_SEs[SEID] + SE = loads(SEstr) + if isinstance(SE, Security_Environment): + self.current_SE = SE + else: + raise SwError(SW["ERR_REFNOTUSABLE"]) + + return SW["NORMAL"], "" + + + def erase_SE(self, SEID): + """ + Erases a Security Environment stored under SEID from the SAM + """ + if (not self.saved_SEs.has_key(SEID)): + raise SwError(SW["ERR_REFNOTUSABLE"]) + else: + del self.saved_SEs[SEID] + + return SW["NORMAL"], "" + def set_asym_algorithm(self, cipher, keytype): """ @param cipher: Public/private key object from used for encryption @@ -123,7 +164,7 @@ class SAM(object): """ logging.debug("Received PIN: %s" % PIN.strip()) - PIN = PIN.replace("\0","") #Strip NULL charakters + PIN = PIN.replace("\0","") #Strip NULL characters if p1 != 0x00: raise SwError(SW["ERR_INCORRECTP1P2"]) @@ -143,7 +184,7 @@ class SAM(object): Change the specified referenced data (e.g. CHV) of the card """ - data = data.replace("\0","") #Strip NULL charakters + data = data.replace("\0","") #Strip NULL characters self.PIN = data return SW["NORMAL"], "" @@ -285,19 +326,19 @@ class SAM(object): #The following commands define the interface to the Secure Messaging functions def generate_public_key_pair(self, p1, p2, data): - return self.SM_handler.generate_public_key_pair(p1, p2, data) + return self.current_SE.generate_public_key_pair(p1, p2, data) def parse_SM_CAPDU(self, CAPDU, header_authentication): - return self.SM_handler.parse_SM_CAPDU(CAPDU, header_authentication) + return self.current_SE.parse_SM_CAPDU(CAPDU, header_authentication) def protect_result(self, sw, unprotected_result): - return self.SM_handler.protect_response(sw, unprotected_result) + return self.current_SE.protect_response(sw, unprotected_result) def perform_security_operation(self, p1, p2, data): - return self.SM_handler.perform_security_operation(p1, p2, data) + return self.current_SE.perform_security_operation(p1, p2, data) def manage_security_environment(self, p1, p2, data): - return self.SM_handler.manage_security_environment(p1, p2, data) + return self.current_SE.manage_security_environment(p1, p2, data) class PassportSAM(SAM): def __init__(self, mf): @@ -314,9 +355,9 @@ class PassportSAM(SAM): self.KSmac = None self.__computeKeys() SAM.__init__(self, None, None, mf) - self.SM_handler = ePass_SM(mf, None, None) - self.SM_handler.current_SE.cct.algorithm = "CC" - self.SM_handler.current_SE.ct.algorithm = "DES3-CBC" + self.current_SE = ePass_SE(mf, None, None) + self.current_SE.cct.algorithm = "CC" + self.current_SE.ct.algorithm = "DES3-CBC" def __computeKeys(self): """ @@ -375,11 +416,11 @@ class PassportSAM(SAM): self.KSmac = self.derive_key(KSseed, 2) #self.ssc = rnd_icc[-4:] + rnd_ifd[-4:] #Set the current SE - self.SM_handler.current_SE.ct.key = self.KSenc - self.SM_handler.current_SE.cct.key = self.KSmac - self.SM_handler.ssc = stringtoint(rnd_icc[-4:] + rnd_ifd[-4:]) - self.SM_handler.current_SE.ct.algorithm = "DES3-CBC" - self.SM_handler.current_SE.cct.algorithm = "CC" + self.current_SE.ct.key = self.KSenc + self.current_SE.cct.key = self.KSmac + self.current_SE.ssc = stringtoint(rnd_icc[-4:] + rnd_ifd[-4:]) + self.current_SE.ct.algorithm = "DES3-CBC" + self.current_SE.cct.algorithm = "CC" return SW["NORMAL"], Eicc + Micc def _mac(self, key, data, ssc = None, dopad=True): @@ -396,10 +437,10 @@ class PassportSAM(SAM): class CryptoflexSAM(SAM): def __init__(self, mf=None): SAM.__init__(self, None, None, mf) - self.SM_handler = CryptoflexSM(mf) + self.current_SE = CryptoflexSE(mf) def generate_public_key_pair(self, p1, p2, data): - asym_key = self.SM_handler.generate_public_key_pair(p1, p2, data) + asym_key = self.current_SE.generate_public_key_pair(p1, p2, data) #TODO: Use SE instead (and remove SAM.set_asym_algorithm) self.set_asym_algorithm(asym_key, 0x07) return SW["NORMAL"], "" @@ -429,7 +470,10 @@ class CryptoflexSAM(SAM): class Security_Environment(object): - def __init__(self): + def __init__(self, MF, SAM): + self.mf = MF + self.sam = SAM + self.SEID = None self.sm_objects = "" @@ -446,20 +490,6 @@ class Security_Environment(object): self.internal_auth = False self.externel_auth = False -class Secure_Messaging(object): - - def __init__(self, MF, SAM, SE=None): - self.mf = MF - self.sam = SAM - - #Security Environments may be saved and retrieved from/to this dictionary - self.saved_SEs = {} - - if not SE: - self.current_SE = Security_Environment() - else: - self.current_SE = SE - def set_MF(self, mf): self.mf = mf @@ -489,23 +519,23 @@ class Secure_Messaging(object): if(cmd == 0x01): #Secure messaging in command data field if se & 0x01: - self.current_SE.capdu_sm = True + self.capdu_sm = True #Secure messaging in response data field if se & 0x02: - self.current_SE.rapdu_sm = True + self.rapdu_sm = True #Computation, decipherment, internal authentication and key agreement if se & 0x04: - self.current_SE.internal_auth = True + self.internal_auth = True #Verification, encipherment, external authentication and key agreement if se & 0x08: - self.current_SE.external_auth = True + self.external_auth = True return self.__set_SE(p2, data) elif(cmd== 0x02): - return self.__store_SE(p2) + return self.sam.store_SE(p2) elif(cmd == 0x03): - return self.__restore_SE(p2) + return self.sam.restore_SE(p2) elif(cmd == 0x04): - return self.__erase_SE(p2) + return self.sam.erase_SE(p2) else: raise SwError(SW["ERR_INCORRECTP1P2"]) @@ -520,57 +550,18 @@ class Secure_Messaging(object): if not p2 in valid_p2: raise SwError(SW["ERR_INCORRECTP1P2"]) if p2 == 0xA4: - return self.current_SE.at.parse_SE_config(data) + return self.at.parse_SE_config(data) elif p2 == 0xA6: - return self.current_SE.kat.parse_SE_config(data) + return self.kat.parse_SE_config(data) elif p2 == 0xAA: - return self.current_SE.ht.parse_SE_config(data) + return self.ht.parse_SE_config(data) elif p2 == 0xB4: - return self.current_SE.cct.parse_SE_config(data) + return self.cct.parse_SE_config(data) elif p2 == 0xB6: - return self.current_SE.dst.parse_SE_config(data) + return self.dst.parse_SE_config(data) elif p2 == 0xB8: - return self.current_SE.ct.parse_SE_config(data) - - def __store_SE(self, SEID): - """ - Stores the current Security environment in the secure access module. The - SEID is used as a reference to identify the SE. - """ - SEstr = dumps(self.current_SE) - self.saved_SEs[SEID] = SEstr - return SW["NORMAL"], "" - - def __restore_SE(self, SEID): - """ - Restores a Security Environment from the SAM and replaces the current SE - with it - """ - - if (not self.saved_SEs.has_key(SEID)): - raise SwError(SW["ERR_REFNOTUSABLE"]) - else: - SEstr = self.saved_SEs[SEID] - SE = loads(SEstr) - if isinstance(SE, Security_Environment): - self.current_SE = SE - else: - raise SwError(SW["ERR_REFNOTUSABLE"]) - - return SW["NORMAL"], "" - - - def __erase_SE(self, SEID): - """ - Erases a Security Environment stored under SEID from the SAM - """ - if (not self.saved_SEs.has_key(SEID)): - raise SwError(SW["ERR_REFNOTUSABLE"]) - else: - del self.saved_SEs[SEID] - - return SW["NORMAL"], "" - + return self.ct.parse_SE_config(data) + def parse_SM_CAPDU(self, CAPDU, header_authentication): """ This methods parses a data field including Secure Messaging objects. @@ -583,7 +574,7 @@ class Secure_Messaging(object): """ structure = TLVutils.unpack(CAPDU.data) return_data = ["",] - expected = self.current_SE.sm_objects + expected = self.sm_objects cla = None ins = None @@ -675,8 +666,7 @@ class Secure_Messaging(object): """ padding_indicator = stringtoint(value[0]) sw, plain = self.decipher(tag, 0x80, value[1:]) - plain = vsCrypto.strip_padding(self.current_SE.ct.algorithm, - plain, + plain = vsCrypto.strip_padding(self.ct.algorithm, plain, padding_indicator) return_data.append(plain) @@ -728,7 +718,7 @@ class Secure_Messaging(object): This method protects a response APDU using secure messaging mechanisms It returns the protected data and the SW bytes """ - expected = self.current_SE.sm_objects + expected = self.sm_objects for pos in range(len(expected)): tag = expected[pos] @@ -750,15 +740,15 @@ class Secure_Messaging(object): return_data += encrypted_tlv if sw == SW["NORMAL"]: - if self.current_SE.cct.algorithm == None: + if self.cct.algorithm == None: raise SwError(SW["CONDITIONSNOTSATISFIED"]) - elif self.current_SE.cct.algorithm == "CCT": + elif self.cct.algorithm == "CCT": tag = SM_Class["CHECKSUM"] to_auth = vsCrypto.append_padding("DES-ECB", return_data) sw, auth = self.compute_cryptographic_checksum(0x8E, 0x80, to_auth) length = len(auth) return_data += TLVutils.pack([(tag, length, auth)]) - elif self.current_SE.cct.algorithm == "SIGNATURE": + elif self.cct.algorithm == "SIGNATURE": tag = SM_Class["DIGITAL_SIGNATURE"] hash = self.hash(0x90, 0x80, return_data) sw, auth = self.compute_digital_signature(0x9E, 0x9A, hash) @@ -811,13 +801,11 @@ class Secure_Messaging(object): """ if p1 != 0x8E or p2 != 0x80: raise SwError(SW["ERR_INCORRECTP1P2"]) - if self.current_SE.cct.key == None: + if self.cct.key == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) - checksum = vsCrypto.crypto_checksum(self.current_SE.cct.algorithm, - self.current_SE.cct.key, - data, - self.current_SE.cct.iv) + checksum = vsCrypto.crypto_checksum(self.cct.algorithm, self.cct.key, + data, self.cct.iv) return SW["NORMAL"], checksum def compute_digital_signature(self, p1, p2, data): @@ -832,7 +820,7 @@ class Secure_Messaging(object): if p1 != 0x9E or not p2 in (0x9A, 0xAC, 0xBC): raise SwError(SW["ERR_INCORRECTP1P2"]) - if self.current_SE.dst.key == None: + if self.dst.key == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) to_sign = "" @@ -846,7 +834,7 @@ class Secure_Messaging(object): elif p2 == 0xBC: #Data objects to be signed pass - signature = self.current_SE.dst.key.sign(to_sign, "") + signature = self.dst.key.sign(to_sign, "") return SW["NORMAL"], signature def hash(self, p1, p2, data): @@ -857,7 +845,7 @@ class Secure_Messaging(object): """ if p1 != 0x90 or not p2 in (0x80, 0xA0): raise SwError(SW["ERR_INCORRECTP1P2"]) - algo = self.current_SE.ht.algorithm + algo = self.ht.algorithm if algo == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) try: @@ -876,9 +864,9 @@ class Secure_Messaging(object): plain = "" cct = "" - algo = self.current_SE.cct.algorithm - key = self.current_SE.cct.key - iv = self.current_SE.cct.iv + algo = self.cct.algorithm + key = self.cct.key + iv = self.cct.iv if algo == None or key == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) @@ -903,7 +891,7 @@ class Secure_Messaging(object): contain a data to sign (tag 0x9A, 0xAC or 0xBC) and a digital signature (0x9E) """ - key = self.current_SE.dst.key + key = self.dst.key to_sign = "" signature = "" @@ -942,13 +930,13 @@ class Secure_Messaging(object): by the current Security environment. Return raw data (no TLV coding). """ - algo = self.current_SE.ct.algorithm - key = self.current_SE.ct.key + algo = self.ct.algorithm + key = self.ct.key if key == None or algo == None: return SW["ERR_CONDITIONNOTSATISFIED"], "" else: padded = vsCrypto.append_padding(algo, data) - crypted = vsCrypto.encrypt(algo, key, padded, self.current_SE.ct.iv) + crypted = vsCrypto.encrypt(algo, key, padded, self.ct.iv) return SW["NORMAL"], crypted def decipher(self, p1, p2, data): @@ -957,12 +945,12 @@ class Secure_Messaging(object): by the current Security environment. Return raw data (no TLV coding). Padding is not removed!!! """ - algo = self.current_SE.ct.algorithm - key = self.current_SE.ct.key + algo = self.ct.algorithm + key = self.ct.key if key == None or algo == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) else: - plain = vsCrypto.decrypt(algo, key, data, self.current_SE.ct.iv) + plain = vsCrypto.decrypt(algo, key, data, self.ct.iv) return SW["NORMAL"], plain def generate_public_key_pair(self, p1, p2, data): @@ -973,16 +961,15 @@ class Secure_Messaging(object): from Crypto.Util.randpool import RandomPool rnd = RandomPool() - cipher = self.current_SE.ct.algorithm + cipher = self.ct.algorithm c_class = locals().get(cipher, None) if c_class is None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) if p1 & 0x01 == 0x00: #Generate key - PublicKey = c_class.generate(self.current_SE.dst.keylength, - rnd.get_bytes) - self.current_SE.dst.key = PublicKey + PublicKey = c_class.generate(self.dst.keylength, rnd.get_bytes) + self.dst.key = PublicKey else: pass #Read key @@ -1016,9 +1003,9 @@ class Secure_Messaging(object): return SW["NORMAL"], "" #}}} -class CryptoflexSM(Secure_Messaging): +class CryptoflexSE(Security_Environment): def __init__(self, mf): - Secure_Messaging.__init__(self, mf) #Does Cryptoflex need its own SE? + Security_Environment.__init__(self, mf) def generate_public_key_pair(self, p1, p2, data): """ @@ -1043,7 +1030,7 @@ class CryptoflexSM(Secure_Messaging): rnd = RandomPool() PublicKey = RSA.generate(keylength, rnd.get_bytes) - self.current_SE.dst.key = PublicKey + self.dst.key = PublicKey e_in = struct.unpack("