added lifecycle commands to CardOS

Passive Authentication for BAC MRTDs


git-svn-id: svn+ssh://localhost/home/henryk/svn/cyberflex-shell/trunk@254 f711b948-2313-0410-aaa9-d29f33439f0b

README

@@ -14,6 +14,13 @@ case.
 You will also need the Crypto module for cryptography support, see 
 http://www.amk.ca/python/code/crypto.html
 
+For Passive Authentication of MRTDs (Passports with BAC) M2Crypto is used, see
+http://wiki.osafoundation.org/bin/view/Projects/MeTooCrypto
+which in turn requires openssl
+http://www.openssl.org
+and swig
+http://www.swig.org
+
 The shell has only been tested on a Linux system but should be platform
 independent, if you have Python, PC/SC and pycsc for your target platform.
 I am however unable to give support for any platform but Linux. 

cards/cardos_card.py

@@ -9,11 +9,23 @@ class CardOS_Card(ISO_7816_4_Card,building_blocks.Card_with_ls):
             ("3bf2180002c10a31fe58c80874", None),
         ]
 
+    APDU_LIFECYCLE = C_APDU("\x00\xCA\x01\x83\x00")
+    APDU_PHASE_CONTROL = C_APDU("\x80\x10\x00\x00\x00")
     APDU_LIST_X = C_APDU("\x80\x16\x01\x00\x00")
     LIST_X_DF = 0
     LIST_X_EF = 1
     LS_L_SIZE_TAG = 0x80
 
+    CARDOS_LIFE_CYCLE_STATUS_BYTE_DESCRIPTIONS = [
+         (0x10, "operational"),
+         (0x20,  "Administration"),
+         (0x23, "Personalization"),
+         (0x26, "Initialisation"),
+         (0x34, "Manufacturing"),
+         (0x3F, "Death"),
+         (0x29, "Erase in Progress"),
+     ]
+
     STATUS_WORDS = ( {
         "6283": "File is deactivated",
         "6300": "Authentication failed",
@@ -82,8 +94,24 @@ def cmd_listfiles(self):
         result = self.list_x(1)
         print "EFs: " + ", ".join([utils.hexdump(a, short=True) for a in result])
 
+    def cmd_lifecycle(self):
+        "Check the current lifecycle"
+        result = self.send_apdu(C_APDU(self.APDU_LIFECYCLE))
+        #status = binascii.b2a_hex(result.data)
+        for hex, mes in self.CARDOS_LIFE_CYCLE_STATUS_BYTE_DESCRIPTIONS:
+          if (int(binascii.b2a_hex(result.data), 16) == hex):
+             print "Satus: " +  mes
+             break
+
+    def cmd_phase_control(self):
+        "change lifecycle between Administration and Operational"
+        result = self.send_apdu(C_APDU(self.APDU_PHASE_CONTROL))
+
+
     COMMANDS = {
         "list_dirs": cmd_listdirs,
         "list_files": cmd_listfiles,
         "ls": building_blocks.Card_with_ls.cmd_list,
+        "check_lifecycle": cmd_lifecycle, 
+        "phase_control": cmd_phase_control,
         }

cards/passport_application.py

@@ -208,6 +208,118 @@ def cmd_perform_bac(self, mrz2, verbose=1):
 
         self.se = Passport_Security_Environment(self)
 
+    def verify_cms(self, data):
+        """Verify a pkcs7 SMIME message"""    
+        from M2Crypto import SMIME, X509, BIO
+        import base64, TLV_utils,  os
+
+        data3 = "-----BEGIN PKCS7-----\n"
+        data3 += base64.encodestring(data)
+        data3 += "-----END PKCS7-----"    
+        #print data3
+
+        p7_bio = BIO.MemoryBuffer(data3)
+
+        # Instantiate an SMIME object.
+        s = SMIME.SMIME()
+
+        # TODO: ugly hack for M2Crypto
+        body = TLV_utils.tlv_find_tag(TLV_utils.unpack(data), 0xA0,  1)[0][2]
+        thecert = TLV_utils.tlv_find_tag(body, 0xA0,  2)[1][2]
+
+        cert_bio = BIO.MemoryBuffer(TLV_utils.pack(thecert))
+
+        # Load the signer's cert.
+        x509 = X509.load_cert_bio(cert_bio,  format=0)
+        sk = X509.X509_Stack()
+        sk.push(x509)
+        s.set_x509_stack(sk)
+        country = str(x509.get_issuer()).split('/')[1][2:]
+        #print country
+
+        cacert = country + "-cacert.der"
+        #print cacert
+
+        msgErr =  "couldn't parse certificate to determine URL for CACert, search the intertubes,"
+        msg =  "download CACert (convert to DER if necessary) and save it as \n\"%s\"" % cacert
+
+        if not os.path.isfile(cacert):
+            try:
+                v = x509.get_ext("certificatePolicies").get_value()
+                start = v.find("CPS: ")
+                if start != -1:
+                    url = v[start + 5:-1]
+                    print "visit %s" % url,  msg
+                else:
+                    print msgErr,  msg
+            except Exception:
+                print msgErr,  msg
+            return ""
+
+        # Load the signer's CA cert. 
+        st = X509.X509_Store()
+        #st.load_info('main')
+        x509CA = X509.load_cert(cacert,  format=0)
+        st.add_x509(x509CA)
+        s.set_x509_store(st)
+
+        # Load the data, verify it.
+        #p7, data = SMIME.smime_load_pkcs7_bio(p7_bio)
+        p7 = SMIME.load_pkcs7_bio(p7_bio)
+
+        v = s.verify(p7)
+        return v
+
+    def cmd_passive_auth(self, verbose=1):
+        "Perform passive authentication"
+
+        hashes = {}
+        result = ""
+        i = 0
+
+        for name in ("DG1",  "DG2",  "SOD"):
+            fid = None
+            for n, f in self.INTERESTING_FILES:
+                if n == name: 
+                    fid = f
+                    break
+            if fid is None:
+                return
+
+            i += 1
+            result = self.open_file(fid, 0x0c)
+            if self.check_sw(result.sw):
+                contents, sw = self.read_binary_file()
+                #self.last_result = R_APDU(contents + self.last_sw)
+
+                if name != "SOD":
+                    hashes[i] = crypto_utils.hash("SHA", contents)
+                else:
+                    result = self.verify_cms(contents[4:])
+
+        #print hexdump(result)
+        #print "DG1: %s" % hexdump(hashes[i])
+        #print "DG2: %s" % hexdump(hashes[2])
+
+        res = TLV_utils.tlv_find_tag(TLV_utils.unpack(result), 0x04)
+        if len(res) == 0:
+            print "failed to verify EF.SOD"
+            return
+        else:
+            print "verified EF.SOD"
+
+        i = 0
+        for tag, length, hash in res:
+            i += 1
+            if hexdump(hashes[i]) == hexdump(hash):
+                print "DG%d hash verified: %s" % (i, binascii.b2a_hex(hash))
+            else:
+                print "DG%d hash failed:" % i
+                print "was:      %s" % binascii.b2a_hex(hashes[i])
+                print "expected: %s" % binascii.b2a_hex(hash)
+                return
+
+
     def before_send(self, apdu):
         if self.se:
             apdu = self.se.before_send(apdu)
@@ -293,6 +405,7 @@ def cmd_read_dg(self, dg):
         "read_dg": cmd_read_dg,
         "parse_biometrics": cmd_parse_biometrics,
         "parse_passport": cmd_parse_passport,
+        "passive_authenticate":cmd_passive_auth, 
     }
 
     DATA_GROUPS = {

crypto_utils.py

@@ -42,6 +42,26 @@ def cipher(do_encrypt, cipherspec, key, data, iv = None):
 
     del cipher
     return result
+
+def hash(hashspec, data):
+    """Do a cryptographic hash operation.
+    hashspec must be of the form "cipher\""""
+    from Crypto.Hash import SHA, RIPEMD, MD2, MD4, MD5
+
+    if len(hashspec) != 3 and len(hashspec) != 6:
+        raise ValueError, 'hashspec must be one of SHA, RIPEMD, MD2, MD4, MD5'
+
+    h_class = locals().get(hashspec.upper(), None)
+    if h_class is None: 
+        raise ValueError, "Hash '%s' not known, must be one of %s" % (hashspec, ", ".join([e.lower() for e in dir() if e.isupper()]))
+
+    hash = h_class.new()        
+    hash.update(data)
+    result = hash.digest()
+    #m.hexdigest()
+
+    del hash
+    return result
 
 def operation_on_string(string1, string2, op):
     if len(string1) != len(string2):