import SmartcardFilesystem, CryptoUtils, TLVutils import random, re, struct, Crypto import hashlib from SWutils import SwError, SW, SW_MESSAGES from pickle import dumps, loads from utils import inttostring, stringtoint, hexdump, C_APDU, R_APDU from Crypto.Cipher import DES3 from ConstantDefinitions import * from SEutils import ControlReferenceTemplate as CRT def generate_SAM_config(cardNumber,PIN,cardSecret,path,password): """ This method generates a new SAM configuration using the given parameters, stores it in a KeyContainer, serialises the container, encrypts it and writes it to the disk. """ container = CardContainer(cardNumber,PIN,cardSecret) container.saveConfiguration(path,password) del container def get_referenced_cipher(p1): """ P1 defines the algorithm and mode to use. We dispatch it and return a string that is understood by CryptoUtils.py functions """ ciphertable = { 0x01: "DES3-ECB", 0x02: "DES3-CBC", 0x03: "DES-ECB", 0x04: "DES-CBC", 0x05: "AES-ECB", 0x06: "AES-CBC", 0x07: "RSA", 0x08: "DSA" } if (ciphertable.has_key(p1)): return ciphertable[p1] else: raise SwError(SW["ERR_INCORRECTP1P2"]) class CardContainer: """ This class is used to store the data needed by the SAM It includes the PIN, the master key of the SAM and a hashmap containing all the keys used by the file encryption system. The keys in the hashmap are indexed via the path to the corresponding container. """ def __init__(self,cardNumber=None,PIN=None,cardSecret=None): self.cardNumber = cardNumber self.PIN = PIN self.cardSecret = cardSecret self.FSkeys = {} self.cipher = 0x01 # Algorithm reference defined in __get_referenced_algorithm(p1) self.block_size = 16 #Need to be set according to the needs of self.cipher. Ugly! self.master_password = None self.master_key = None self.salt = None self.asym_key = None def __delete__(self): print "Smartcard configuration is NOT saved!!!" def getPIN(self): return self.PIN def setPIN(self,PIN): self.PIN = PIN def getCardSecret(self): return self.cardSecret def addKey(self,path,key): """ Encrypt key and store it in the SAM """ crypted_key = CryptoUtils.cipher(True,self.cipher,self.master_password,key) if self.FSkeys.has_key(path): print "Overwriting key for path %s" % path self.FSkeys[path] = crypted_key def removeKey(self,path): """ Erase a key from the SAM """ if(self.FSkeys.has_key(path)): del self.FSkeys[path] else: raise ValueError def getKey(self,path): """ Fetch a key from the SAM, decrypt and return it @return: key if the key is in the container, otherwise None """ if(self.FSkeys.has_key(path)): plain_key = CryptoUtils.cipher(False,self.cipher, self.master_password,self.FSkeys[path]) #TODO: Error checking: return plain_key else: return None def loadConfiguration(self,path,password): """ Reads the configuration of a key container from the disk, decrypts it and applies it to the current key_ container. The current configuration gets overwritten. The HMAC is used to verify data integrity. """ decryptedContainer = self.loadFromDisk(path, password, True) container = loads(decryptedContainer) self.PIN = container.PIN self.cardNumber = container.cardNumber self.cardSecret = container.cardSecret self.FSkeys = container.FSkeys self.master_password = password del decryptedContainer del container def saveConfiguration(self,path,password=None): """ Encrypts the configuration of the current key container and writes it to the disk. Password is used to derivate a key using PBKDF2. A salt and a HMAC is stored along with the data """ serialisedContainer = dumps(self) if self.master_key == None: #New Configuration if password == None: raise ValueError, "Need password but none specified" pbk = CryptoUtils.crypt(password) regex = re.compile('\$p5k2\$\$[\w./]*\$') match = regex.match(pbk) if match != None: self.salt = pbk[7:match.end()-1] self.master_key = pbk[match.end():] else: raise ValueError del pbk self.saveToDisk(serialisedContainer,path) def saveToDisk(self,data,path,password=None): if password == None: key = self.master_key salt = self.salt else: regex = re.compile('\$p5k2\$\$[\w./]*\$') pbk = CryptoUtils.crypt(password) match = regex.match(pbk) salt = pbk[7:match.end()-1] print salt key = pbk[match.end():] cipher = get_referenced_cipher(self.cipher) paddedData = CryptoUtils.append_padding(cipher,data) cryptedData = CryptoUtils.cipher(True,cipher,key,paddedData) hmac = CryptoUtils.crypto_checksum("HMAC",key,cryptedData) data = "$p5k2$$" + salt + "$" + cryptedData + hmac try: fd = open(path,"w") fd.write(data) fd.close() except IOError: print "Failed to write data to disk" def loadFromDisk(self,path,password=None,loading_sam=False): hmac_length = 32 #FIXME: Ugly #Read data try: fd = open(path,"r") data = fd.read() fd.close() except IOError: print "Failed to read data from disk" #Check file structure, ectract components regex = re.compile('\$p5k2\$\$[\w./]*\$') #TODO: Ensure the right format! match = regex.match(data) if match != None: crypted = data[match.end():len(data) - hmac_length] salt = data[7:match.end() - 1] hmac = data[len(data) - hmac_length:] else: raise ValueError, "Wrong file format" if password == None: if salt != self.salt: raise ValueError, "Could not decode data" else: key = self.master_key else: pbk = CryptoUtils.crypt(password,salt) match = regex.match(pbk) key = pbk[match.end():] #Verify HMAC checksum = CryptoUtils.crypto_checksum("HMAC",key,crypted) if checksum != hmac: print "Found HMAC %s expected %s" % (str(hmac), str(checksum)) raise ValueError, "Failed to authenticate data. Wrong password?" #Decrypt data cipher = get_referenced_cipher(self.cipher) decrypted = CryptoUtils.cipher(False,cipher,key,crypted) #If we are loading the SAM config (paramter loading_sam) we set #the master key and the salt #This is ugly and should be changed if loading_sam: self.master_key = key self.salt = salt return decrypted class SAM(object): def __init__(self,path,password,mf=None): """ Reads the encrypted SAM configuration from the disk and applies it. """ self.CardContainer = CardContainer() if len(password) % self.CardContainer.block_size != 0: print "Wrong key length, must be a multiple of %s Bytes" % self.CardContainer.block_size raise ValueError self.CardContainer.loadConfiguration(path, password) self.mf = mf self.card_number = self.CardContainer.cardNumber self.last_challenge = None #Will contain non-readable binary string self.counter = 3 #Number of tries for PIN validation self.cardSecret = self.CardContainer.cardSecret self.SM_handler = Secure_Messaging(self.mf) def set_MF(self,mf): self.mf = mf self.SM_handler.set_MF(mf) def addFSkey(self,path,key): #Deprecated ? self.CardContainer.addKey(path, key) def delFSkey(self,path): #Deprecated ? self.CardContainer.removeKey(path) def FSencrypt(self,data): """ Encrypt the given data, using the parameters stored in the SAM """ if self.CardContainer.master_key == None: raise ValueError cipher = get_referenced_cipher(self.CardContainer.cipher) padded_data = CryptoUtils.append_padding(cipher,data) crypted_data = CryptoUtils.cipher(True,cipher,self.CardContainer.master_key,padded_data) return crypted_data def FSdecrypt(self,data): """ Decrypt the given data, using the parameters stored in the SAM """ if self.CardContainer.master_key == None: raise ValueError, "No master key set." cipher = get_referenced_cipher(self.CardContainer.cipher) decrypted_data = CryptoUtils.cipher(False,cipher,self.CardContainer.master_key,data) unpadded_data = CryptoUtils.strip_padding(cipher,decrypted_data) return unpadded_data def set_algorithm(self,algo): if not algo in range(0x01,0x06): raise ValueError, "Illegal Parameter" else: self.CardContainer.cipher = algo def set_asym_algorithm(self,cipher,type): """ @param cipher: Public/private key object from Crypto.PublicKey used for encryption @param type: Type of the public key (e.g. RSA, DSA) """ if not type in range(0x07,0x08): raise ValueError, "Illegal Parameter" else: self.CardContainer.cipher = type self.CardContainer.asym_key = cipher def verify(self,p1,p2,PIN): """ Authenticate the card user. Check if he entered a valid PIN. If the PIN is invalid decrement retry counter. If retry counter equals zero, block the card until reset with correct PUK """ print "Received PIN: %s" % PIN.strip() PIN = PIN.replace("\0","") #Strip NULL charakters if p1 != 0x00: raise SwError(SW["ERR_INCORRECTP1P2"]) if self.counter > 0: if self.CardContainer.getPIN() == PIN: self.counter = 3 return SW["NORMAL"], "" else: self.counter -= 1 raise SwError(SW["WARN_NOINFO63"]) #raise SwError(0X63C0 + self.counter) #Be verbose else: raise SwError(SW["ERR_AUTHBLOCKED"]) def change_reference_data(self,p1,p2,data): """ Change the specified referenced data (e.g. CHV) of the card """ data = data.replace("\0","") #Strip NULL charakters self.CardContainer.setPIN(data) print "New PIN = %s" % self.CardContainer.getPIN() return SW["NORMAL"], "" def internal_authenticate(self,p1,p2,data): """ Authenticate card to terminal. Encrypt the challenge of the terminal to prove key posession """ if p1 == 0x00: #No information given cipher = get_referenced_cipher(self.CardContainer.cipher) else: cipher = get_referenced_cipher(p1) if cipher == "RSA" or cipher == "DSA": crypted_challenge = self.CardContainer.asym_key.sign(data,"") crypted_challenge = crypted_challenge[0] crypted_challenge = inttostring(crypted_challenge) else: key = self._get_referenced_key(p1,p2) crypted_challenge = CryptoUtils.cipher(True,cipher,key,data) return SW["NORMAL"], crypted_challenge def external_authenticate(self,p1,p2,data): """ Authenticate the terminal to the card. Check wether Terminal correctly encrypted the given challenge or not """ if self.last_challenge is None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) key = self._get_referenced_key(p1,p2) if p1 == 0x00: #No information given cipher = get_referenced_cipher(self.CardContainer.cipher) else: cipher = get_referenced_cipher(p1) reference = CryptoUtils.append_padding(cipher,self.last_challenge) reference = CryptoUtils.cipher(True,cipher,key,reference) if(reference == data): #Invalidate last challenge self.last_challenge = None return SW["NORMAL"], "" else: plain = CryptoUtils.cipher(False,cipher,key,data) print plain print "Reference: " + hexdump(reference) print "Data: " + hexdump(data) raise SwError(SW["WARN_NOINFO63"]) #TODO: Counter for external authenticate? def mutual_authenticate(self,p1,p2,mutual_challenge): """ Takes an encrypted challenge in the form 'Terminal Challenge | Card Challenge | Card number' and checks it for validity. If the challenge is succesful the card encrypts 'Card Challenge | Terminal challenge' and returns this value """ key = self._get_referenced_key(p1,p2) card_number = self.get_card_number() if (key == None): raise SwError(SW["ERR_INCORRECTP1P2"]) if p1 == 0x00: #No information given cipher = get_referenced_cipher(self.CardContainer.cipher) else: cipher = get_referenced_cipher(p1) if (cipher == None): raise SwError(SW["ERR_INCORRECTP1P2"]) plain = CryptoUtils.cipher(False,cipher,key,mutual_challenge) terminal_challenge = plain[:len(self.last_challenge)-1] card_challenge = plain[len(self.last_challenge):len(challenge)-len(card_number)-1] serial_number = plain[(challenge)-len(card_number):] if terminal_challenge != self.last_challenge: print "Mutual authentication failed!" raise SwError(SW["WARN_NOINFO63"]) elif serial_number != card_number: print "Mutual authentication failed!" raise SwError(SW["WARN_NOINFO63"]) result = card_challenge + terminal_challenge return SW["NORMAL"], CryptoUtils.cipher(True,cipher,key,result) def get_challenge(self,p1,p2,data): """ Generate a random number of maximum 8 Byte and return it. """ if (p1 != 0x00 or p2 != 0x00): #RFU raise SwError(SW["ERR_INCORRECTP1P2"]) length = 8 #Length of the challenge in Byte max = "0x" + length * "ff" #cipher = get_referenced_cipher(self.CardContainer.cipher) #block_size = Crypto.Cipher.cipher.block_size random.seed() self.last_challenge = random.randint(0, int(max,16)) print "Generated challenge: %s" % str(self.last_challenge) self.last_challenge = inttostring(self.last_challenge,length) return SW["NORMAL"], self.last_challenge def get_card_number(self): return SW["NORMAL"], inttostring(self.card_number) def _get_referenced_key(self,p1,p2): """ This method returns the key specified by the p2 parameter. The key may be stored on the cards filesystem or in memory. @param p1: Specifies the algorithm to use. Needed to know the keylength. @param p2: Specifies a reference to the key to be used for encryption """ """ Meaning of p2: b8 b7 b6 b5 b4 b3 b2 b1 | Meaning 0 0 0 0 0 0 0 0 | No information is given 0 -- -- -- -- -- -- -- | Global reference data (e.g. an MF secific key) 1 -- -- -- -- -- -- -- | Specific reference data (e.g. DF specific key) -- -- -- x x x x x | Number of the secret Any other value | RFU """ key = None qualifier = p2 & 0x1F algo = get_referenced_cipher(p1) cipher = CryptoUtils.get_cipher(p1) keylength = cipher.key_size if (p2 == 0x00): #No information given, use the global card key key = self.CardContainer.cardSecret #elif ((p2 >> 7) == 0x01 or (p2 >> 7) == 0x00): #We treat global and specific reference data alike else: try: #Interpret qualifier as an short fid and try to read the key from the FS if self.mf == None: raise SwError(SW["CONDITIONSNOTSATISFIED"]) df = self.mf.currentDF() fid = df.select("fid", stringtoint(qualifier)) key = fid.readbinary(keylength) except SwError: key = None if key == None: #Try to read the key from the Key Container key = self.CardContainer.getKey(reference_data) if key != None: return key else: raise SwError(SW["DATANOTFOUND"]) def _read_FS_data(self,fid,length): """ This function creates a dummy filesystem, so that data (for example keys) can be read from the FS. It is here for testing purposes. Later on the OS interface to the FS should be used. """ mf = SmartcardFilesystem.MF(filedescriptor=FDB["NOTSHAREABLEFILE"], lifecycle=LCB["NOINFORMATION"], dfname=None) mf = SmartcardFilesystem.MF(filedescriptor=FDB["DF"],lifecycle=LCB["ACTIVATED"], dfname=None) df = mf.append(SmartcardFilesystem.DF(parent=mf,fid=4, dfname='\xd2\x76\x00\x01\x24\x01', bertlv_data=[])) mf.append(SmartcardFilesystem.TransparentStructureEF(parent=df, fid=0x1012, filedescriptor=0, data="Key from the FS.")) df = mf.currentDF() file = df.select('fid', 0X1012) sw, result = file.readbinary(0) return sw, result #The following commands define the interface to the CardContainer functions def saveConfiguration(self,path,password=None): return self.CardContainer.saveConfiguration(path, password) def loadConfiguration(self,path,password): return self.CardContainer.loadConfiguration(path, password) def loadFromDisk(self,path,password=None): return self.CardContainer.loadFromDisk(path, password) def saveToDisk(self,data,path,password=None): return self.CardContainer.saveToDisk(data, path, password) #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) def parse_SM_CAPDU(self,CAPDU,header_authentication): return self.SM_handler.parse_SM_CAPDU(CAPDU, header_authentication) def protect_result(self,sw,unprotected_result): return self.SM_handler.protect_response(sw, unprotected_result) def perform_security_operation(self,p1,p2,data): return self.SM_handler.perform_security_operation(p1, p2, data) def manage_security_environment(self,p1,p2,data): return self.SM_handler.manage_security_environment(p1, p2, data) class PassportSAM(SAM): #MRZ1 = "Pi", c) H = hashlib.sha1(D).digest() Ka = H[0:8] Kb = H[8:16] return Ka + Kb def external_authenticate(self,p1,p2,resp_data): """Performes the basic access controll protocoll as defined in the ICAO MRTD standard""" rnd_icc = self.last_challenge #Receive Mutual Authenticate APDU from terminal #Decrypt data and check MAC #cipher = DES3.new(self.KEnc,"des3-cbc") Eifd = resp_data[:-8] Mifd = self._mac(self.KMac, Eifd) #Check the MAC if not Mifd == resp_data[-8:]: raise ValueError, "Passport authentication failed: Wrong MAC on incoming data during Mutual Authenticate" #Decrypt the data plain = CryptoUtils.cipher(False,"DES3-CBC",self.KEnc,resp_data[:-8]) #Split decrypted data into the two nonces and if plain[8:16] != rnd_icc: raise SwError(SW["WARN_NOINFO63"]) #Extraxt keying material from IFD, generate ICC keying material Kifd = plain[16:] rnd_ifd = plain[:8] max = "0x" + 16 * "ff" random.seed() Kicc = inttostring(random.randint(0, int(max,16))) #Generate Answer data = plain[8:16] + plain[:8] + Kicc Eicc = CryptoUtils.cipher(True,"DES3-CBC",self.KEnc,data) Micc = self._mac(self.KMac,Eicc) #Derive the final keys KSseed = CryptoUtils.operation_on_string(Kicc, Kifd, lambda a,b: a^b) self.KSenc = self.derive_key(KSseed, 1) self.KSmac = self.derive_key(KSseed, 2) #self.ssc = rnd_icc[-4:] + rnd_ifd[-4:] #Set the current SE self.SM_handler.CAPDU_SE.ct.key = self.KSenc self.SM_handler.CAPDU_SE.cct.key = self.KSmac self.SM_handler.RAPDU_SE.ct.key = self.KSenc self.SM_handler.RAPDU_SE.cct.key = self.KSmac self.SM_handler.ssc = stringtoint(rnd_icc[-4:] + rnd_ifd[-4:]) self.SM_handler.CAPDU_SE.ct.algo = "DES3-CBC" self.SM_handler.RAPDU_SE.ct.algo = "DES3-CBC" self.SM_handler.CAPDU_SE.cct.algo = "CC" self.SM_handler.RAPDU_SE.cct.algo = "CC" return SW["NORMAL"], Eicc + Micc def _mac(self, key, data, ssc = None, dopad=True): if ssc: data = ssc + data if dopad: topad = 8 - len(data) % 8 data = data + "\x80" + ("\x00" * (topad-1)) a = CryptoUtils.cipher(True, "des-cbc", key[:8], data) b = CryptoUtils.cipher(False, "des-ecb", key[8:16], a[-8:]) c = CryptoUtils.cipher(True, "des-ecb", key[:8], b) return c class CryptoflexSAM(SAM): def __init__(self,path="testconfig.sam",key="DUMMYKEYDUMMYKEY",mf=None): SAM.__init__(self,path,key,mf) self.SM_handler = CryptoflexSM(mf) def generate_public_key_pair(self,p1,p2,data): asym_key = self.SM_handler.generate_public_key_pair(p1, p2, data) self.set_asym_algorithm(asym_key, 0x07) return SW["NORMAL"], "" def perform_security_operation(self,p1,p2,data): """ In the cryptoflex card, this is the verify key command. A key is send to the card in plain text and compared to a key stored in the card. This is used for authentication @param data: Contains the key to be verified @return: SW[NORMAL] in case of success otherwise SW[WARN_NOINFO63] """ return SW["NORMAL"], "" #FIXME #key = self._get_referenced_key(p1,p2) #if key == data: # return SW["NORMAL"], "" #else: # return SW["WARN_NOINFO63"], "" def internal_authenticate(self,p1,p2,data): data = data[::-1] #Reverse Byte order sw, data = SAM.internal_authenticate(self, p1, p2, data) if data != "": data = data[::-1] return sw, data class Security_Environment(object): def __init__(self): self.SEID = None self.sm_objects = "" #Default configuration ct = "\x80\x01\x01\x82\x01\x00\x85\x01\x00" #Control Reference Tables self.at = CRT(TEMPLATE_AT) self.kat = CRT(TEMPLATE_KAT) self.ht = CRT(TEMPLATE_HT) self.cct = CRT(TEMPLATE_CCT) self.dst = CRT(TEMPLATE_DST) self.ct = CRT(TEMPLATE_CT) def mse(self,config): structure = TLVutils.unpack(config) for tag, length, value in structure: if tag == TEMPLATE_AT: self.at.parse_config(config) elif tag == TEMPLATE_CRT: self.crt.parse_config(config) elif tag == TEMPLATE_CCT: self.cct.parse_config(config) elif tag == TEMPLATE_KAT: self.kat.parse_config(config) elif tag == TEMPLATE_CT: self.ct.parse_config(config) elif tag == TEMPLATE_DST: self.dst.parse_config(config) else: raise ValueError class Secure_Messaging(object): def __init__(self,MF,CAPDU_SE=None,RAPDU_SE=None): self.mf = MF if not CAPDU_SE: self.CAPDU_SE = Security_Environment() else: self.CAPDU_SE = CAPDU_SE if not RAPDU_SE: self.RAPDU_SE = Security_Environment() else: self.RAPDU_SE = RAPDU_SE self.current_SE = Security_Environment() def set_MF(self,mf): self.mf = mf def manage_security_environment(self,p1,p2,data): """ This method is used to store, restore or erase Security Environments or to manipualte the various parameters of the current SE. P1 specifies the operation to perform, p2 is either the SEID for the referred SE or the tag of a control reference template """ """ P1: b8 b7 b6 b5 b4 b3 b2 b1 Meaning - - - 1 - - - - Secure messaging in command data field - - 1 - - - - - Secure messaging in response data field - 1 - - - - - - Computation, decipherment, internal authentication and key agreement 1 - - - - - - - Verification, encipherment, external authentication and key agreement - - - - 0 0 0 1 SET 1 1 1 1 0 0 1 0 STORE 1 1 1 1 0 0 1 1 RESTORE 1 1 1 1 0 1 0 0 ERASE """ cmd = p1 & 0x0F se = p1 & 0xF0 if(cmd == 0x01): if se == 0x01: #Set the right SE self.current_SE = self.CAPDU_SE elif se == 0x02: self.current_SE = self.RAPDU_SE else: pass #FIXME self.__set_SE(p2,data) elif(cmd== 0x02): self.__store_SE(p2) elif(cmd == 0x03): self.__restore_SE(p2) elif(cmd == 0x04): self.__erase_SE(p2) else: raise SwError(SW["ERR_INCORRECTP1P2"]) def __set_SE(self,p2,data): """ Manipulate the current Security Environment. P2 is the tag of a control reference template, data contains control reference objects """ valid_p2 = (0xA4,0xA6,0xB4,0xB6,0xB8) if not p2 in valid_p2: raise SwError(SW["ERR_INCORRECTP1P2"]) tag, length, value = TLVutils.unpack(data) #Only one object? if p2 == 0xA4: self.current_SE.at.parse_SE_config(value) elif p2 == 0xA6: self.current_SE.kat.parse_SE_config(value) elif p2 == 0xAA: self.current_SE.ht.parse_SE_config(value) elif p2 == 0xB4: self.current_SE.cct.parse_SE_config(value) elif p2 == 0xB6: self.current_SE.dst.parse_SE_config(value) elif p2 == 0xB8: self.current_SE.ct.parse_SE_config(value) 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.CAPDU_SE) try: self.SAM.addkey(SEID, SEstr) #TODO: Need SAM reference except ValueError: raise SwError["ERR_INCORRECTP1P2"] def __restore_SE(self,SEID): """ Restores a Security Environment from the SAM and replaces the current SE with it """ SEstr = self.SAM.get_key(SEID) SE = loads(SEstr) if isinstance(SE, SecurityEnvironment): self.CAPDU_SE = SE else: raise ValueError def __erase_SE(self,SEID): """ Erases a Security Environment stored under SEID from the SAM """ self.SAM.removeKey(SEID) def parse_SM_CAPDU(self,CAPDU,header_authentication): """ Frontend for the __parse_SM command. We set the right Security Environment and restore the old one, when the command is finished """ current_SE = self.current_SE self.current_SE = self.CAPDU_SE try: capdu = self.__parse_SM(CAPDU,header_authentication) except SwError, e: #Restore Security Environment self.current_SE = current_SE raise e self.current_SE = current_SE return capdu def __parse_SM(self,CAPDU,header_authentication): """ This methods parses a data field including Secure Messaging objects. SM_header indicates wether or not the header of the message shall be authenticated It returns an unprotected command APDU @param CAPDU: The protected CAPDU to be parsed @param header_authentication: Wether or not the header should be included in authentication mechanisms @return: Unprotected command APDU """ structure = TLVutils.unpack(CAPDU.data) return_data = ["",]; expected = self.current_SE.sm_objects cla = None ins = None p1 = None p2 = None le = None if header_authentication: to_authenticate = inttostring(CAPDU.cla) + inttostring(CAPDU.ins) + inttostring(CAPDU.p1) + inttostring(CAPDU.p2) to_authenticate = CryptoUtils.append_padding("DES-CBC",to_authenticate,0x01,8) else: to_authenticate = "" for object in structure: tag, length, value = object #TODO: Sanity checking #if not SM_Class.has_key(tag): # raise ValueError, "unknown Secure messaging tag %#X" % tag if tag % 2 == 1: #Include object in checksum calculation to_authenticate += inttostring(tag) + inttostring(length) + value #SM data objects for encapsulating plain values if tag in (SM_Class["PLAIN_VALUE_NO_TLV"],SM_Class["PLAIN_VALUE_NO_TLV_ODD"]): return_data.append(value) #FIXME: Need TLV coding? elif tag in (SM_Class["PLAIN_VALUE_TLV_INCULDING_SM"],SM_Class["PLAIN_VALUE_TLV_INCULDING_SM_ODD"]): #Encapsulated SM objects. Parse them return_data.append(self.parse_SM_CAPDU(value, header_authentication)) #FIXME: Need to pack value into a dummy CAPDU elif tag in (SM_Class["PLAIN_VALUE_TLV_NO_SM"],SM_Class["PLAIN_VALUE_TLV_NO_SM_ODD"]): #Encapsulated plaintext BER-TLV objects return_data.append(value) elif tag in (SM_Class["Ne"],SM_Class["Ne_ODD"]): le = value elif tag == SM_Class["PLAIN_COMMAND_HEADER"]: if len(value) != 8: raise ValueError, "Plain Command Header expected, but received insufficient data" else: cla = value[:2] ins = value[2:4] p1 = value[4:6] p2 = value[6:8] #SM data objects for confidentiality if tag in (SM_Class["CRYPTOGRAM_PLAIN_TLV_INCLUDING_SM"],SM_Class["CRYPTOGRAM_PLAIN_TLV_INCLUDING_SM_ODD"]): #The Cryptogram includes SM objects. We decrypt them and parse the objects. plain = decipher(tag,0x80,value) #TODO: Need Le = length return_data.append(self.parse_SM_CAPDU(plain, header_authentication)) elif tag in (SM_Class["CRYPTOGRAM_PLAIN_TLV_NO_SM"],SM_Class["CRYPTOGRAM_PLAIN_TLV_NO_SM_ODD"]): #The Cryptogram includes BER-TLV enconded plaintext. We decrypt them and return the objects. plain = decipher(tag,0x80,value) return_data.append(plain) elif tag in (SM_Class["CRYPTOGRAM_PADDING_INDICATOR"],SM_Class["CRYPTOGRAM_PADDING_INDICATOR_ODD"]): #The first byte of the specified data field indicates the padding to use: """ Value Meaning '00' No further indication '01' Padding as specified in 6.2.3.1 '02' No padding '1X' One to four secret keys for enciphering information, not keys ('X' is a bitmap with any value from '0' to 'F') '11' indicates the first key (e.g., an "even" control word in a pay TV system) '12' indicates the second key (e.g., an "odd" control word in a pay TV system) '13' indicates the first key followed by the second key (e.g., a pair of control words in a pay TV system) '2X' Secret key for enciphering keys, not information ('X' is a reference with any value from '0' to 'F') (e.g., in a pay TV system, either an operational key for enciphering control words, or a management key for enciphering operational keys) '3X' Private key of an asymmetric key pair ('X' is a reference with any value from '0' to 'F') '4X' Password ('X' is a reference with any value from '0' to 'F') '80' to '8E' Proprietary """ padding_indicator = stringtoint(value[0]) sw, plain = self.decipher(tag,0x80,value[1:]) if sw != 0x9000: raise ValueError plain = CryptoUtils.strip_padding(self.CAPDU_SE.ct.algo,plain,padding_indicator) return_data.append(plain) #SM data objects for authentication if tag == SM_Class["CHECKSUM"]: auth = CryptoUtils.append_padding("DES-CBC",to_authenticate,0x01,8) sw, checksum = self.compute_cryptographic_checksum(0x8E, 0x80, auth) if checksum != value: print "Failed to verify checksum!" raise SwError(SW["ERR_SECMESSOBJECTSINCORRECT"]) elif tag == SM_Class["DIGITAL_SIGNATURE"]: auth = to_authenticate #FIXME: Need padding? sw, signature = self.compute_digital_signature(0x9E, 0x9A, auth) if signature != value: print "Failed to verify signature!" raise SwEroor(SW["ERR_SECMESSOBJECTSINCORRECT"]) elif tag in (SM_Class["HASH_CODE"],SM_Class["HASH_CODE_ODD"]): sw, hash = self.hash(p1, p2, to_authenticate) if hash != value: print "Failed to verify hash!" raise SwEroor(SW["ERR_SECMESSOBJECTSINCORRECT"]) #Check if we just parsed a expected SM Object: pos = 0 while (pos < len (expected)): if expected[pos] == tag: expected = expected[:pos-1] + expected[pos:] break pos += 1 #Form unprotected CAPDU if cla == None: cla = CAPDU.cla if ins == None: ins = CAPDU.ins if p1 == None: p1 = CAPDU.p1 if p2 == None: p2 = CAPDU.p2 if le == None: le = CAPDU.le if expected != "": raise SwError(SW["ERR_SECMESSOBJECTSMISSING"]) c = C_APDU(cla=cla,ins=ins,p1=p1,p2=p2,le=le,data="".join(return_data)) return c def protect_response(self,sw,result): """ Frontend for the __protect command. We set the right Security Environment and restore the old one, when the command is finished """ current_SE = self.current_SE self.current_SE = self.RAPDU_SE try: sw, data = self.__protect(sw,result) except SwError, e: #Restore Security Environment self.current_SE = current_SE raise e self.current_SE = current_SE return sw, data def __protect(self,sw,result): """ This method protects a response APDU using secure messanging mechanisms It returns the protected data and the SW bytes """ expected = self.current_SE.sm_objects for pos in range(len(expected)): tag = expected[pos] return_data = "" #if sw == SW["NORMAL"]: # sw = inttostring(sw) # length = len(sw) # tag = SM_Class["PLAIN_PROCESSING_STATUS"] # tlv_sw = TLVutils.pack([(tag,length,sw)]) # return_data += tlv_sw if result != "": # Encrypt the data included in the RAPDU sw, encrypted = self.encipher(0x82, 0x80, result) encrypted = "\x01" + encrypted encrypted_tlv = TLVutils.pack([(SM_Class["CRYPTOGRAM_PADDING_INDICATOR_ODD"],len(encrypted),encrypted)]) return_data += encrypted_tlv if sw == SW["NORMAL"]: if self.CAPDU_SE.cct.algo == None: raise SwError(SW["CONDITIONSNOTSATISFIED"]) elif self.CAPDU_SE.cct.algo == "CCT": tag = SM_Class["CHECKSUM"] to_auth = CryptoUtils.append_padding("DES-ECB",return_data,0x01,8) sw, auth = self.compute_cryptographic_checksum(0x8E, 0x80, to_auth) length = len(auth) return_data += TLVutils.pack([(tag,length,auth)]) elif self.CAPDU_SE.cct.algo == "SIGNATURE": tag = SM_Class["DIGITAL_SIGNATURE"] hash = self.hash(0x90, 0x80, return_data) sw, auth = self.compute_digital_signature(0x9E, 0x9A, hash) length = len(auth) return_data += TLVutils.pack([(tag,length,auth)]) return SW["NORMAL"], return_data #The following commands implement ISO 7816-8 {{{ def perform_security_operation(self,p1,p2,data): """ In the end this command is nothing but a big switch for all the other commands in ISO 7816-8. It will invoke the appropiate command and return its result """ allowed_P1P2 = ((0x90,0x80),(0x90,0xA0),(0x9E,0x9A),(0x9E,0xAC),(0x9E,0xBC),(0x00,0xA2), (0x00,0xA8),(0x00,0x92),(0x00,0xAE),(0x00,0xBE),(0x82,0x80),(0x84,0x80), (0x86,0x80),(0x80,0x82),(0x80,0x84),(0x80,0x86)) if (p1,p2) not in allowed_P1P2: raise SwError(SW["INCORRECTP1P2"]) if(p2 in (0x80,0xA0) and p1==0x90): sw, response_data = self.hash(p1,p2,data) elif(p2 in (0x9A,0xAC,0xBC) and p1 == 0x9E): sw, response_data = self.compute_digital_signature(p1, p2, data) elif(p2 == 0xA2 and p1 == 0x00): sw, response_data = self.verify_cryptographic_checksum(p1, p2, data) elif(p2 == 0xA8 and p1 == 0x00): sw, response_data = self.verify_digital_signature(p1, p2, data) elif(p2 in (0x92,0xAE,0xBE) and p1 == 0x00): sw, response_data = self.verify_certificate(p1, p2, data) elif (p2 == 0x80 and p1 in (0x82,0x84,0x86)): sw, response_data = self.encipher(p1, p2, data) elif (p2 in (0x82,0x84,0x86) and p1 == 0x80): sw, response_data = self.decipher(p1, p2, data) if p1 == 0x00: assert response_data == "" return sw, response_data def compute_cryptographic_checksum(self,p1,p2,data): """ Compute a cryptographic checksum (e.g. MAC) for the given data. Algorithm and key are specified in the current (CAPDU) SE @bug: Always use CAPDU settings ??? """ if p1 != 0x8E or p2 != 0x80: raise SwError(SW["ERR_INCORRECTP1P2"]) if self.CAPDU_SE.cct.algo == None or self.current_SE.cct.key == None: raise SWError(SE["ERR_CONDITIONNOTSATISFIED"]) checksum = CryptoUtils.crypto_checksum(self.current_SE.cct.algo, self.current_SE.cct.key, data, self.current_SE.cct.iv) return SW["NORMAL"], checksum def compute_digital_signature(self,p1,p2,data): """ Compute a digital signature for the given data. Algorithm and key are specified in the current SE @param p1: Must be 0x9E = Secure Messaging class for digital signatures @param p2: Must be one of 0x9A, 0xAC, 0xBC. Indicates what kind of data is included in the data field. """ if p1 != 0x9E or not p2 in (0x9A,0xAC,0xBC): raise SwError(SW["ERR_INCORRECTP1P2"]) if self.current_SE.dst.key == None: raise SWError(SE["ERR_CONDITIONNOTSATISFIED"]) if p2 == 0x9A: #Data to be signed to_sign = data elif p2 == 0xAC: #Data objects, sign values to_sign = "" structure = TLVutils.unpack(data) for tag, length, value in structure: to_sign += value elif p2 == 0xBC: #Data objects to be signed pass signature = self.current_SE.dst.key.sign(data,"") return SW["NORMAL"], signature def hash(self,p1,p2,data): """ Hash the given data using the algorithm specified by the current Security environment. Return raw data (no TLV coding). """ if p1 != 0x90 or not p2 in (0x80,0xA0): raise SwError(SW["ERR_INCORRECTP1P2"]) algo = self.current_SE.ht.algo if algo == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) try: hash = CryptoUtils.hash("MD5",data) except ValueError: #FIXME: Type of error raise SwError(SW["ERR_SECMESSNOTSUPPORTED"]) return SW["NORMAL"], hash def verify_cryptographic_checksum(self,p1,p2,data): """ Verify the cryptographic checksum contained in the data field. Data field must contain a cryptographic checksum (tag 0x8E) and a plain value (tag 0x80) """ plain = "" cct = "" algo = self.current_SE.cct.algo key = self.current_SE.cct.key iv = self.current_SE.cct.iv if algo == None or key == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) structure = TLVutils.unpack(config) for tag, length, value in structure: if tag == 0x80: plain = value elif tag == 0x8E: cct = value if plain == "" or cct == "": raise SwError(SW["ERR_SECMESSOBJECTSMISSING"]) else: my_cct = CryptoUtils.crypto_checksum(algo,key,plain,iv) if my_cct == cct: return SW["NORMAL"], "" else: raise SwError["ERR_SECMESSOBJECTSINCORRECT"] def verify_digital_signature(self,p1,p2,data): """ Verify the digital signature contained in the data field. Data must contain a data to sign (tag 0x9A, 0xAC or 0xBC) and a digital signature (0x9E) """ key = self.current_SE.dst.key to_sign = "" signature = "" if key == None: raise SwError(SW["ERR_CONDITIONNOTSATISFIED"]) structure = TLVutils.unpack(config) for tag, length, value in structure: if tag == 0x9E: signature = value elif tag == 0x9A: #FIXME: Correct treatment of all possible tags to_sign = value elif tag == 0xAC: pass elif tag == 0xBC: pass if to_sign == "" or signature == "": raise SwError(SW["ERR_SECMESSOBJECTSMISSING"]) try: my_signature = key.sign(value) except error, e: #FIXME: Foobar pass if my_signature == signature: return SW["NORMAL"], "" else: raise SwError(["ERR_SECMESSOBJECTSINCORRECT"]) def verify_certificate(self,p1,p2,data): if p1 != 0x00 or p2 not in (0x92,0xAE,0xBE): raise SwError(SW["ERR_INCORRECTP1P2"]) pass def encipher(self,p1,p2,data): """ Encipher data using key, algorithm, IV and Padding specified by the current Security environment. Return raw data (no TLV coding). """ algo = self.current_SE.ct.algo key = self.current_SE.ct.key if key == None or algo == None: return SW["ERR_CONDITIONNOTSATISFIED"], "" else: padded = CryptoUtils.append_padding(algo,data) crypted = CryptoUtils.cipher(True,algo,key,padded,self.current_SE.ct.iv) return SW["NORMAL"], crypted def decipher(self,p1,p2,data): """ Decipher data using key, algorithm, IV and Padding specified by the current Security environment. Return raw data (no TLV coding). Padding is not removed!!! """ algo = self.current_SE.ct.algo key = self.current_SE.ct.key if key == None or algo == None: return SW["ERR_CONDITIONNOTSATISFIED"], "" else: plain = CryptoUtils.cipher(False,algo,key,data,self.current_SE.ct.iv) return SW["NORMAL"], plain def generate_public_key_pair(self,p1,p2,data): """ Generate a new public-private key pair. """ from Crypto.PublicKey import RSA, DSA from Crypto.Util.randpool import RandomPool rnd = RandomPool() cipher = self.CAPDU_SE.ct.algo #FIXME: Current SE? 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.CAPDU_SE.dst.keylength,rnd.get_bytes) self.CAPDU_SE.dst.key = PublicKey else: pass #Read key #Encode keys if cipher == "RSA": #Public key n = str(PublicKey.__getstate__()['n']) e = str(PublicKey.__getstate__()['e']) pk = ((0x81,len(n),n),(0x82,len(e),e)) result = TLVutils.bertlv_pack(pk) #result = TLVutils.bertlv_pack((0x7F49,len(pk),pk)) #Private key d = PublicKey.__getstate__()['d'] elif cipher == "DSA": #Public key p = str(PublicKey.__getstate__()['p']) q = str(PublicKey.__getstate__()['q']) g = str(PublicKey.__getstate__()['g']) y = str(PublicKey.__getstate__()['y']) #Private key x = str(PublicKey.__getstate__()['x']) #Add more algorithms here if p1 & 0x02 == 0x02: return SW["NORMAL"], result else: #FIXME: Where to put the keys? return SW["NORMAL"], "" #}}} class CryptoflexSM(Secure_Messaging): def __init__(self,mf): Secure_Messaging.__init__(self,mf) #Does Cryptoflex need its own SE? def generate_public_key_pair(self,p1,p2,data): """ In the Cryptoflex card this command only supports RSA keys. @param data: Contains the public exponent used for key generation @param p1: The keynumber. Can be used later to refer to the generated key @param p2: Used to specifiy the keylength. The mapping is: 0x40 => 256 Bit, 0x60 => 512 Bit, 0x80 => 1024 """ from Crypto.PublicKey import RSA from Crypto.Util.randpool import RandomPool keynumber = p1 #TODO: Check if key exists keylength_dict = {0x40: 256, 0x60: 512, 0x80: 1024} if not keylength_dict.has_key(p2): raise SwError(SW["ERR_INCORRECTP1P2"]) else: keylength = keylength_dict[p2] rnd = RandomPool() PublicKey = RSA.generate(keylength,rnd.get_bytes) self.CAPDU_SE.dst.key = PublicKey e_in = struct.unpack("