diff --git a/virtualsmartcard/src/vpicc/virtualsmartcard/SEutils.py b/virtualsmartcard/src/vpicc/virtualsmartcard/SEutils.py
index 73c13a8..7ef5295 100644
--- a/virtualsmartcard/src/vpicc/virtualsmartcard/SEutils.py
+++ b/virtualsmartcard/src/vpicc/virtualsmartcard/SEutils.py
@@ -17,11 +17,14 @@
# virtualsmartcard. If not, see .
#
import TLVutils
+from virtualsmartcard.ConstantDefinitions import CRT_TEMPLATE, ALGO_MAPPING, SM_Class
+from virtualsmartcard.utils import inttostring, stringtoint, C_APDU
+from virtualsmartcard.SWutils import SwError, SW
+import virtualsmartcard.CryptoUtils as vsCrypto
+
+import struct, logging
from time import time
from random import seed, randint
-from virtualsmartcard.ConstantDefinitions import CRT_TEMPLATE, ALGO_MAPPING
-from virtualsmartcard.utils import inttostring, stringtoint
-from virtualsmartcard.SWutils import SwError, SW
class ControlReferenceTemplate:
"""
@@ -154,3 +157,620 @@ class ControlReferenceTemplate:
return self.__config_string
+class Security_Environment(object):
+
+ def __init__(self, MF, SAM):
+ self.mf = MF
+ self.sam = SAM
+
+ self.SEID = None
+ self.sm_objects = ""
+
+ #Control Reference Tables
+ self.at = ControlReferenceTemplate(CRT_TEMPLATE["AT"])
+ self.kat = ControlReferenceTemplate(CRT_TEMPLATE["KAT"])
+ self.ht = ControlReferenceTemplate(CRT_TEMPLATE["HT"])
+ self.cct = ControlReferenceTemplate(CRT_TEMPLATE["CCT"])
+ self.dst = ControlReferenceTemplate(CRT_TEMPLATE["DST"])
+ self.ct = ControlReferenceTemplate(CRT_TEMPLATE["CT"])
+
+ self.capdu_sm = False
+ self.rapdu_sm = False
+ self.internal_auth = False
+ self.externel_auth = False
+
+ 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 manipulate 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 >> 4
+ if(cmd == 0x01):
+ #Secure messaging in command data field
+ if se & 0x01:
+ self.capdu_sm = True
+ #Secure messaging in response data field
+ if se & 0x02:
+ self.rapdu_sm = True
+ #Computation, decipherment, internal authentication and key agreement
+ if se & 0x04:
+ self.internal_auth = True
+ #Verification, encipherment, external authentication and key agreement
+ if se & 0x08:
+ self.external_auth = True
+ return self.__set_SE(p2, data)
+ elif(cmd== 0x02):
+ return self.sam.store_SE(p2)
+ elif(cmd == 0x03):
+ return self.sam.restore_SE(p2)
+ elif(cmd == 0x04):
+ return self.sam.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"])
+ if p2 == 0xA4:
+ return self.at.parse_SE_config(data)
+ elif p2 == 0xA6:
+ return self.kat.parse_SE_config(data)
+ elif p2 == 0xAA:
+ return self.ht.parse_SE_config(data)
+ elif p2 == 0xB4:
+ return self.cct.parse_SE_config(data)
+ elif p2 == 0xB6:
+ return self.dst.parse_SE_config(data)
+ elif p2 == 0xB8:
+ 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.
+ SM_header indicates whether 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.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 = vsCrypto.append_padding("DES-CBC", to_authenticate)
+ else:
+ to_authenticate = ""
+
+ for tlv in structure:
+ tag, length, value = tlv
+
+ 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?
+ #Encapsulated SM objects. Parse them
+ #FIXME: Need to pack value into a dummy CAPDU
+ elif tag in (SM_Class["PLAIN_VALUE_TLV_INCULDING_SM"],
+ SM_Class["PLAIN_VALUE_TLV_INCULDING_SM_ODD"]):
+ return_data.append(self.parse_SM_CAPDU(value, header_authentication))
+ #Encapsulated plaintext BER-TLV objects
+ elif tag in (SM_Class["PLAIN_VALUE_TLV_NO_SM"],
+ SM_Class["PLAIN_VALUE_TLV_NO_SM_ODD"]):
+ 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 SwError(SW["ERR_SECMESSOBJECTSINCORRECT"])
+ 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 = self.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 = self.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 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:])
+ plain = vsCrypto.strip_padding(self.ct.algorithm, plain,
+ padding_indicator)
+ return_data.append(plain)
+
+ #SM data objects for authentication
+ if tag == SM_Class["CHECKSUM"]:
+ auth = vsCrypto.append_padding("DES-CBC", to_authenticate)
+ sw, checksum = self.compute_cryptographic_checksum(0x8E,
+ 0x80,
+ auth)
+ if checksum != value:
+ 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:
+ raise SwError(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:
+ raise SwError(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):
+ """
+ This method protects a response APDU using secure messaging mechanisms
+ It returns the protected data and the SW bytes
+ """
+ expected = self.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.cct.algorithm == None:
+ raise SwError(SW["CONDITIONSNOTSATISFIED"])
+ 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.cct.algorithm == "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 appropriate 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 SE
+ """
+ if p1 != 0x8E or p2 != 0x80:
+ raise SwError(SW["ERR_INCORRECTP1P2"])
+ if self.cct.key == None:
+ raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
+
+ 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):
+ """
+ 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.dst.key == None:
+ raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
+
+ to_sign = ""
+ 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.dst.key.sign(to_sign, "")
+ 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.ht.algorithm
+ if algo == None:
+ raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
+ try:
+ hash = vsCrypto.hash(algo, data)
+ except ValueError:
+ raise SwError(SW["ERR_EXECUTION"])
+
+ 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.cct.algorithm
+ key = self.cct.key
+ iv = self.cct.iv
+ if algo == None or key == None:
+ raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
+
+ structure = TLVutils.unpack(data)
+ 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 = vsCrypto.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.dst.key
+ to_sign = ""
+ signature = ""
+
+ if key == None:
+ raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
+
+ structure = TLVutils.unpack(data)
+ 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"])
+
+ my_signature = key.sign(value)
+ if my_signature == signature:
+ return SW["NORMAL"], ""
+ else:
+ raise SwError(["ERR_SECMESSOBJECTSINCORRECT"])
+
+ def verify_certificate(self, p1, p2, data):
+ """
+ Verify a certificate send by the terminal using the internal trust
+ anchors.
+ This method is currently not implemented.
+ """
+ if p1 != 0x00 or p2 not in (0x92, 0xAE, 0xBE):
+ raise SwError(SW["ERR_INCORRECTP1P2"])
+ else:
+ raise NotImplementedError
+
+ 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.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.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.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.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.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.dst.keylength, rnd.get_bytes)
+ self.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":
+ #DSAParams
+ p = str(PublicKey.__getstate__()['p'])
+ q = str(PublicKey.__getstate__()['q'])
+ g = str(PublicKey.__getstate__()['g'])
+ #Public key
+ y = str(PublicKey.__getstate__()['y'])
+ #TODO: Actual encoding
+ #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 CryptoflexSE(Security_Environment):
+ def __init__(self, mf):
+ Security_Environment.__init__(self, mf)
+
+ 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 specify 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.dst.key = PublicKey
+
+ e_in = struct.unpack("> 4
- if(cmd == 0x01):
- #Secure messaging in command data field
- if se & 0x01:
- self.capdu_sm = True
- #Secure messaging in response data field
- if se & 0x02:
- self.rapdu_sm = True
- #Computation, decipherment, internal authentication and key agreement
- if se & 0x04:
- self.internal_auth = True
- #Verification, encipherment, external authentication and key agreement
- if se & 0x08:
- self.external_auth = True
- return self.__set_SE(p2, data)
- elif(cmd== 0x02):
- return self.sam.store_SE(p2)
- elif(cmd == 0x03):
- return self.sam.restore_SE(p2)
- elif(cmd == 0x04):
- return self.sam.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"])
- if p2 == 0xA4:
- return self.at.parse_SE_config(data)
- elif p2 == 0xA6:
- return self.kat.parse_SE_config(data)
- elif p2 == 0xAA:
- return self.ht.parse_SE_config(data)
- elif p2 == 0xB4:
- return self.cct.parse_SE_config(data)
- elif p2 == 0xB6:
- return self.dst.parse_SE_config(data)
- elif p2 == 0xB8:
- 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.
- SM_header indicates whether 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.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 = vsCrypto.append_padding("DES-CBC", to_authenticate)
- else:
- to_authenticate = ""
-
- for tlv in structure:
- tag, length, value = tlv
-
- 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?
- #Encapsulated SM objects. Parse them
- #FIXME: Need to pack value into a dummy CAPDU
- elif tag in (SM_Class["PLAIN_VALUE_TLV_INCULDING_SM"],
- SM_Class["PLAIN_VALUE_TLV_INCULDING_SM_ODD"]):
- return_data.append(self.parse_SM_CAPDU(value, header_authentication))
- #Encapsulated plaintext BER-TLV objects
- elif tag in (SM_Class["PLAIN_VALUE_TLV_NO_SM"],
- SM_Class["PLAIN_VALUE_TLV_NO_SM_ODD"]):
- 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 SwError(SW["ERR_SECMESSOBJECTSINCORRECT"])
- 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 = self.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 = self.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 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:])
- plain = vsCrypto.strip_padding(self.ct.algorithm, plain,
- padding_indicator)
- return_data.append(plain)
-
- #SM data objects for authentication
- if tag == SM_Class["CHECKSUM"]:
- auth = vsCrypto.append_padding("DES-CBC", to_authenticate)
- sw, checksum = self.compute_cryptographic_checksum(0x8E,
- 0x80,
- auth)
- if checksum != value:
- 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:
- raise SwError(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:
- raise SwError(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):
- """
- This method protects a response APDU using secure messaging mechanisms
- It returns the protected data and the SW bytes
- """
- expected = self.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.cct.algorithm == None:
- raise SwError(SW["CONDITIONSNOTSATISFIED"])
- 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.cct.algorithm == "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 appropriate 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 SE
- """
- if p1 != 0x8E or p2 != 0x80:
- raise SwError(SW["ERR_INCORRECTP1P2"])
- if self.cct.key == None:
- raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
-
- 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):
- """
- 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.dst.key == None:
- raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
-
- to_sign = ""
- 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.dst.key.sign(to_sign, "")
- 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.ht.algorithm
- if algo == None:
- raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
- try:
- hash = vsCrypto.hash(algo, data)
- except ValueError:
- raise SwError(SW["ERR_EXECUTION"])
-
- 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.cct.algorithm
- key = self.cct.key
- iv = self.cct.iv
- if algo == None or key == None:
- raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
-
- structure = TLVutils.unpack(data)
- 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 = vsCrypto.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.dst.key
- to_sign = ""
- signature = ""
-
- if key == None:
- raise SwError(SW["ERR_CONDITIONNOTSATISFIED"])
-
- structure = TLVutils.unpack(data)
- 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"])
-
- my_signature = key.sign(value)
- if my_signature == signature:
- return SW["NORMAL"], ""
- else:
- raise SwError(["ERR_SECMESSOBJECTSINCORRECT"])
-
- def verify_certificate(self, p1, p2, data):
- """
- Verify a certificate send by the terminal using the internal trust
- anchors.
- This method is currently not implemented.
- """
- if p1 != 0x00 or p2 not in (0x92, 0xAE, 0xBE):
- raise SwError(SW["ERR_INCORRECTP1P2"])
- else:
- raise NotImplementedError
-
- 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.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.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.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.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.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.dst.keylength, rnd.get_bytes)
- self.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":
- #DSAParams
- p = str(PublicKey.__getstate__()['p'])
- q = str(PublicKey.__getstate__()['q'])
- g = str(PublicKey.__getstate__()['g'])
- #Public key
- y = str(PublicKey.__getstate__()['y'])
- #TODO: Actual encoding
- #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 CryptoflexSE(Security_Environment):
- def __init__(self, mf):
- Security_Environment.__init__(self, mf)
-
- 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 specify 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.dst.key = PublicKey
-
- e_in = struct.unpack("