VivoKey/vsmartcard
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Moved Security_Environment code to the SEutils file

Browse Source 
git-svn-id: https://vsmartcard.svn.sourceforge.net/svnroot/vsmartcard@497 96b47cad-a561-4643-ad3b-153ac7d7599c
This commit is contained in:
oepen
2011-08-17 14:59:57 +00:00
parent 8d14525349
commit d4b54a9410
2 changed files with 625 additions and 625 deletions
Download Patch File Download Diff File Expand all files Collapse all files

626
virtualsmartcard/src/vpicc/virtualsmartcard/SEutils.py
View File

@@ -17,11 +17,14 @@
# virtualsmartcard. If not, see <http://www.gnu.org/licenses/>.
#
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("<i", data)
if e_in[0] != 65537:
logging.warning("Warning: Exponents different from 65537 are ignored!" +\
"The Exponent given is %i" % e_in[0])
#Encode Public key
n = PublicKey.__getstate__()['n']
n_str = inttostring(n)
n_str = n_str[::-1]
e = PublicKey.__getstate__()['e']
e_str = inttostring(e, 4)
e_str = e_str[::-1]
pad = 187 * '\x30' #We don't have CRT components, so we need to pad
pk_n = TLVutils.bertlv_pack(((0x81, len(n_str), n_str),
(0x01, len(pad), pad),
(0x82, len(e_str), e_str)))
#Private key
d = PublicKey.__getstate__()['d']
#Write result to FID 10 12 EF-PUB-KEY
df = self.mf.currentDF()
ef_pub_key = df.select("fid", 0x1012)
ef_pub_key.writebinary([0], [pk_n])
data = ef_pub_key.getenc('data')
#Write private key to FID 00 12 EF-PRI-KEY (not necessary?)
#How to encode the private key?
ef_priv_key = df.select("fid", 0x0012)
ef_priv_key.writebinary([0], [inttostring(d)])
data = ef_priv_key.getenc('data')
return PublicKey
class ePass_SE(Security_Environment):
def __init__(self, MF, SE, ssc=None):
self.ssc = ssc
Security_Environment.__init__(self, MF, SE)
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. The ePass
uses a Send Sequence Counter for MAC calculation
"""
if p1 != 0x8E or p2 != 0x80:
raise SwError(SW["ERR_INCORRECTP1P2"])
self.ssc += 1
checksum = vsCrypto.crypto_checksum(self.cct.algorithm, self.cct.key,
data, self.cct.iv, self.ssc)
return SW["NORMAL"], checksum

624
virtualsmartcard/src/vpicc/virtualsmartcard/SmartcardSAM.py
View File

@@ -21,12 +21,10 @@ import struct, hashlib, logging
from pickle import dumps, loads
from os import urandom
import TLVutils
import virtualsmartcard.CryptoUtils as vsCrypto
from virtualsmartcard.SWutils import SwError, SW
from virtualsmartcard.utils import inttostring, stringtoint, hexdump, C_APDU
from virtualsmartcard.ConstantDefinitions import SM_Class, CRT_TEMPLATE
from virtualsmartcard.SEutils import ControlReferenceTemplate as CRT
from virtualsmartcard.utils import inttostring, stringtoint
from virtualsmartcard.SEutils import Security_Environment, CryptoflexSE, ePass_SE
def get_referenced_cipher(p1):
"""
@@ -467,624 +465,6 @@ class CryptoflexSAM(SAM):
data = data[::-1]
return sw, data
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 = CRT(CRT_TEMPLATE["AT"])
self.kat = CRT(CRT_TEMPLATE["KAT"])
self.ht = CRT(CRT_TEMPLATE["HT"])
self.cct = CRT(CRT_TEMPLATE["CCT"])
self.dst = CRT(CRT_TEMPLATE["DST"])
self.ct = CRT(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("<i", data)
if e_in[0] != 65537:
logging.warning("Warning: Exponents different from 65537 are ignored!" +\
"The Exponent given is %i" % e_in[0])
#Encode Public key
n = PublicKey.__getstate__()['n']
n_str = inttostring(n)
n_str = n_str[::-1]
e = PublicKey.__getstate__()['e']
e_str = inttostring(e, 4)
e_str = e_str[::-1]
pad = 187 * '\x30' #We don't have CRT components, so we need to pad
pk_n = TLVutils.bertlv_pack(((0x81, len(n_str), n_str),
(0x01, len(pad), pad),
(0x82, len(e_str), e_str)))
#Private key
d = PublicKey.__getstate__()['d']
#Write result to FID 10 12 EF-PUB-KEY
df = self.mf.currentDF()
ef_pub_key = df.select("fid", 0x1012)
ef_pub_key.writebinary([0], [pk_n])
data = ef_pub_key.getenc('data')
#Write private key to FID 00 12 EF-PRI-KEY (not necessary?)
#How to encode the private key?
ef_priv_key = df.select("fid", 0x0012)
ef_priv_key.writebinary([0], [inttostring(d)])
data = ef_priv_key.getenc('data')
return PublicKey
class ePass_SE(Security_Environment):
def __init__(self, MF, SE, ssc=None):
self.ssc = ssc
Security_Environment.__init__(self, MF, SE)
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. The ePass
uses a Send Sequence Counter for MAC calculation
"""
if p1 != 0x8E or p2 != 0x80:
raise SwError(SW["ERR_INCORRECTP1P2"])
self.ssc += 1
checksum = vsCrypto.crypto_checksum(self.cct.algorithm, self.cct.key,
data, self.cct.iv, self.ssc)
return SW["NORMAL"], checksum
if __name__ == "__main__":
"""
Unit test: