Hash for revocation in scan_dcc.py

Author: chrloch

tests/scan_dcc.py

@@ -10,13 +10,17 @@
 from base45 import b45decode
 from cose.messages import Sign1Message
 from datetime import datetime
+from hashlib import sha256
+from base64 import b64encode
 
 def main(args):
     if args.algorithm:
-        with_all_dccs( print_algorithm, error_handler=_throw, exclude='venv*' )
+        with_all_dccs( print_algorithm, error_handler=_throw )
     if not args.validity_at is None:
         validation_clock = datetime.fromisoformat(args.validity_at)
-        with_all_dccs( lambda f: validity_at(validation_clock, f), error_handler=_throw, exclude='venv*' )
+        with_all_dccs( lambda f: validity_at(validation_clock, f), error_handler=_throw )
+    if args.hash:
+        with_all_dccs( print_hashes, error_handler=_display )
 
 def with_all_dccs( function, error_handler=None, exclude="" ):
     ''' Walks through the current directory and all subdirectories and calls
@@ -29,13 +33,12 @@ def with_all_dccs( function, error_handler=None, exclude="" ):
                 fqfn = str(Path(base, file)) # fully qualified file name
                 try: 
                     if not fnmatch( fqfn, exclude ):
-                        function(fqfn )
+                        function( fqfn )
                 except Exception as error:
                     if not error_handler is None: 
                         error_handler( fqfn, error)
 
 
-
 def load_dcc( file ):    
     '''Load QR code from image file and return Sign1Message and Payload
        Usage: s1msg, payload = load_dcc('my_dcc.png')
@@ -54,10 +57,6 @@ def load_dcc_with_rawdata( file ):
     logging.debug('Decoding/Decompressing Base45 data')
     decompressed = zlib.decompress(b45decode(qr_code_data[4:]))
     s1msg = Sign1Message.decode(decompressed)
-    #    print(cbor2.loads(decompressed))
-    #    print(f'Unprotected Header: {cose_data.uhdr}')
-    #    print(f'Protected Header: {cose_data.phdr}')
-    #    print(f'KID = {get_kid_b64(cose_data)}')
     payload = cbor2.loads(s1msg.payload)
     return s1msg, payload, qr_code_data
 
@@ -69,6 +68,36 @@ def _throw( file, error ):
     print('Error reading', file)
     raise error
 
+def _display( file, error ):
+    print('Error reading', file)
+
+
+def get_hashes(file):
+    def hashfunc( value ):
+        'First 16 bytes only, base64-encoded'
+        if isinstance( value, str ):
+            value = value.encode('utf-8')
+        return b64encode(sha256(value).digest()[:16]).decode('utf-8')
+
+
+    s1msg, payload = load_dcc(file)
+    country_code = payload[1]
+    inner = payload[-260][1]
+    uci = inner['v' if 'v' in inner.keys() else 't' if 't' in inner.keys() else 'r'][0]['ci']
+    if get_algorithm(s1msg) == 'ES256':
+        signature = s1msg.signature[:len(s1msg.signature)//2]
+    else:
+        signature = s1msg.signature
+    
+    return {
+        'UCI' : hashfunc(uci),
+        'COUNTRYCODEUCI' : hashfunc(country_code+uci),
+        'SIGNATURE' : hashfunc(signature)
+    }
+
+def print_hashes(file):
+    print(f'{file}\t{get_hashes(file)}')
+
 def validity_at( validation_clock, file ):
     s1msg, payload = load_dcc(file)
     dcc_from = datetime.fromtimestamp(payload[6])
@@ -78,22 +107,24 @@ def validity_at( validation_clock, file ):
     print( '\t'.join([file, validity, validation_clock.isoformat()]))  
 
 
-def print_algorithm( file ):
+def get_algorithm( s1msg ):
     '''Print the algorithm of a DCC 
        ES256 = SHA256 with ECDSA
        PS256 = RSASSA-PSS using SHA-256
     '''
-    s1msg, payload = load_dcc(file)
     for key,value in s1msg.phdr.items(): 
         _key = key.fullname
         if _key == 'ALG':
-            _value = value.fullname
-            print( '\t'.join([file, _value]))  
-
+            return value.fullname
+            
 
+def print_algorithm( file ):
+    s1msg, payload = load_dcc(file)
+    print( '\t'.join([file, get_algorithm(s1msg)]))  
 
 if __name__ == '__main__':
     parser = ArgumentParser(description='Scan all DCCs for something')
+    parser.add_argument('--hash', action='store_true', help='Print hashes')
     parser.add_argument('--algorithm', action='store_true', help='Print algorithm')
     parser.add_argument('--validity-at', action='store', default=None, help='Check validity at ISO date')
     args = parser.parse_args()