RfidResearchGroup#239, RfidResearchGroup#231, RfidResearchGroup#201, RfidResearchGroup#193, and RfidResearchGroup#96(thread locking issue should have been fixed) merged

Author: zricote

CHANGELOG.md

@@ -3,6 +3,8 @@ All notable changes to this project will be documented in this file.
 This project uses the changelog in accordance with [keepchangelog](http://keepachangelog.com/). Please use this to write notable changes, which is not the same as git commit log...
 
 ## [unreleased][unreleased]
+ - Enhanced Nested attack with additional parity bit check to reduce possible keys (@20321587)
+ - `hf mf elog --decrypt` skip records with found keys (@taichunmin)
  - Added `firmware/docker-compose.yml` to build firmware in local docker (@taichunmin)
  - Added cmd to acquire nonces for hardnested(Protocol doc need update) (@xianglin1998)
  - Added command to check keys of multiple sectors at once (@taichunmin)

firmware/application/src/rfid/reader/hf/mf1_toolbox.c

@@ -861,6 +861,7 @@ static uint8_t nested_recover_core(mf1_nested_core_t *pnc, uint64_t keyKnown, ui
     pnc->par |= ((oddparity8(answer[0]) != parity[0]) << 0);
     pnc->par |= ((oddparity8(answer[1]) != parity[1]) << 1);
     pnc->par |= ((oddparity8(answer[2]) != parity[2]) << 2);
+    pnc->par |= ((oddparity8(answer[3]) != parity[3]) << 3);
     return STATUS_HF_TAG_OK;
 }
 

software/script/chameleon_cli_main.py

@@ -68,7 +68,7 @@ def get_prompt(self):
 
         :return: current cmd prompt
         """
-        device_string = f"{CG}USB" if self.device_com.isOpen(
+        device_string = f"{CG}USB" if self.device_com.is_open(
         ) else f"{CR}Offline"
         status = f"[{device_string}{C0}] chameleon --> "
         return status

software/script/chameleon_cli_unit.py

@@ -180,7 +180,7 @@ class DeviceRequiredUnit(BaseCLIUnit):
     """
 
     def before_exec(self, args: argparse.Namespace):
-        ret = self.device_com.isOpen()
+        ret = self.device_com.is_open()
         if ret:
             return True
         else:
@@ -519,13 +519,17 @@ def on_exec(self, args: argparse.Namespace):
 class HWConnect(BaseCLIUnit):
     def args_parser(self) -> ArgumentParserNoExit:
         parser = ArgumentParserNoExit()
-        parser.description = 'Connect to chameleon by serial port'
-        parser.add_argument('-p', '--port', type=str, required=False)
+        parser.description = 'Connect to chameleon by serial port or TCP'
+        parser.add_argument('-p', '--port', type=str, required=False, help='Serial port name/path')
+        parser.add_argument('-t', '--tcp', type=str, required=False, help='TCP server to connect to (host:port)')
         return parser
 
+    def before_exec(self, args: argparse.Namespace):
+        return True
+
     def on_exec(self, args: argparse.Namespace):
         try:
-            if args.port is None:  # Chameleon auto-detect if no port is supplied
+            if args.port is None and args.tcp is None:  # Chameleon auto-detect if no port is supplied
                 platform_name = uname().release
                 if 'Microsoft' in platform_name:
                     path = os.environ["PATH"].split(os.pathsep)
@@ -555,9 +559,13 @@ def on_exec(self, args: argparse.Namespace):
                             args.port = port.device
                             break
                 if args.port is None:  # If no chameleon was found, exit
-                    print("Chameleon not found, please connect the device or try connecting manually with the -p flag.")
+                    print("Chameleon not found, please connect the device or try connecting manually with the -p or -t flag.")
                     return
-            self.device_com.open(args.port)
+            
+            if args.tcp:
+                self.device_com.open(chameleon_com.ChameleonTCPTransport(args.tcp))
+            else:
+                self.device_com.open(chameleon_com.ChameleonSerialTransport(args.port))
             self.device_com.commands = self.cmd.get_device_capabilities()
             major, minor = self.cmd.get_app_version()
             model = ['Ultra', 'Lite'][self.cmd.get_device_model()]
@@ -1449,6 +1457,189 @@ def on_exec(self, args: argparse.Namespace):
         print(f"( {CR}0{C0}: Failed, {CG}1{C0}: Success )\n\n")
 
 
+@hf_mf.command('dump')
+class HFMFDump(MF1AuthArgsUnit):
+    def args_parser(self) -> ArgumentParserNoExit:
+        parser = ArgumentParserNoExit()
+        parser.description = 'Mifare Classic dump tag'
+        parser.add_argument('-t', '--dump-file-type', type=str, required=False, 
+                         help="Dump file content type", choices=['bin', 'hex'])
+        parser.add_argument('-f', '--dump-file', type=argparse.FileType("wb"), required=True,
+                         help="Dump file to write data from tag")
+        parser.add_argument('-d', '--dic', type=argparse.FileType("r"), required=True,
+                         help="Read keys (to communicate with tag to dump) from .dic format file")
+        return parser
+
+    def on_exec(self, args: argparse.Namespace):
+        # check dump type
+        if args.dump_file_type is None:
+            if args.dump_file.name.endswith('.bin'):
+                content_type = 'bin'
+            elif args.dump_file.name.endswith('.eml'):
+                content_type = 'hex'
+            else:
+                raise Exception("Unknown file format, Specify content type with -t option")
+        else:
+            content_type = args.dump_file_type
+
+        # read keys from file
+        keys = [bytes.fromhex(line.strip()) for line in args.dic if line.strip()]
+
+        # data to write to dump file
+        buffer = bytearray()
+
+        # iterate over sectors
+        for s in range(16):
+            # try all keys for this sector
+            typ = None
+            key_found = None
+            for key in keys:
+                # first try key B
+                try:
+                    self.cmd.mf1_read_one_block(4*s, MfcKeyType.B, key)
+                    typ = MfcKeyType.B
+                    key_found = key
+                    break
+                except UnexpectedResponseError:
+                    # ignore read errors at this stage as we want to try key A
+                    pass
+                # try with key A if B was unsuccessful
+                try:
+                    self.cmd.mf1_read_one_block(4*s, MfcKeyType.A, key)
+                    typ = MfcKeyType.A
+                    key_found = key
+                    break
+                except UnexpectedResponseError:
+                    pass
+            else:
+                raise Exception(f"No key found for sector {s}")
+            
+            # iterate over blocks
+            for b in range(4):
+                try:
+                    block_data = self.cmd.mf1_read_one_block(4*s + b, typ, key_found)
+                    # add data to buffer
+                    if content_type == 'bin':
+                        buffer.extend(block_data)
+                    elif content_type == 'hex':
+                        buffer.extend(block_data.hex().encode("utf-8"))
+                except Exception as e:
+                    print(f"Error reading block {4*s + b}: {e}")
+                    # Fill with zeros if we can't read the block
+                    if content_type == 'bin':
+                        buffer.extend(bytes(16))
+                    else:
+                        buffer.extend(b'0' * 32)
+        
+        # write buffer to file
+        args.dump_file.write(buffer)
+        args.dump_file.close()
+        print(f"Dump completed and saved to {args.dump_file.name}")
+
+
+@hf_mf.command('clone')
+class HFMFClone(MF1AuthArgsUnit):
+    def args_parser(self) -> ArgumentParserNoExit:
+        parser = ArgumentParserNoExit()
+        parser.description = 'Mifare Classic clone tag from dump'
+        parser.add_argument('-t', '--dump-file-type', type=str, required=False, 
+                         help="Dump file content type", choices=['bin', 'hex'])
+        parser.add_argument('-a', '--clone-access', action='store_true', 
+                         help="Write ACL from original dump (use with caution, could brick your tag)")
+        parser.add_argument('-f', '--dump-file', type=argparse.FileType("rb"), required=True,
+                         help="Dump file containing data to write on new tag")
+        parser.add_argument('-d', '--dic', type=argparse.FileType("r"), required=True,
+                         help="Read keys (to communicate with tag to write) from .dic format file")
+        return parser
+
+    def on_exec(self, args: argparse.Namespace):
+        # check dump type
+        if args.dump_file_type is None:
+            if args.dump_file.name.endswith('.bin'):
+                content_type = 'bin'
+            elif args.dump_file.name.endswith('.eml'):
+                content_type = 'hex'
+            else:
+                raise Exception("Unknown file format, Specify content type with -t option")
+        else:
+            content_type = args.dump_file_type
+
+        # read data from dump file
+        if content_type == 'bin':
+            buffer = bytearray(args.dump_file.read())
+        else:  # hex
+            buffer = bytearray.fromhex(args.dump_file.read().decode())
+            
+        if len(buffer) % 16 != 0:
+            raise Exception("Data block not aligned to 16 bytes")
+        if len(buffer) // 16 > 256:
+            raise Exception("Data block memory overflow")
+
+        # read keys from file
+        keys = [bytes.fromhex(line.strip()) for line in args.dic if line.strip()]
+
+        # iterate over sectors
+        for s in range(16):
+            # try all keys for this sector
+            keyA, keyB = None, None
+            for key in keys:
+                # first try key B
+                try:
+                    self.cmd.mf1_read_one_block(4*s, MfcKeyType.B, key)
+                    keyB = key
+                except UnexpectedResponseError:
+                    pass
+                # try with key A if B was unsuccessful
+                try:
+                    self.cmd.mf1_read_one_block(4*s, MfcKeyType.A, key)
+                    keyA = key
+                except UnexpectedResponseError:
+                    pass
+                # both keys were found, no need to continue iterating
+                if keyA and keyB:
+                    break
+            
+            if not keyA and not keyB:
+                print(f"Warning: No key found for sector {s}, skipping")
+                continue
+
+            # iterate over blocks
+            for b in range(4):
+                block_data = buffer[(4*s+b)*16:(4*s+b+1)*16]
+                if not block_data or len(block_data) != 16:
+                    print(f"Warning: Invalid block data for block {4*s+b}, skipping")
+                    continue
+
+                # special case for last block of each sector (sector trailer)
+                if b == 3 and not args.clone_access:
+                    # if option is not specified, use generic ACL to be able to write again
+                    block_data = block_data[:6] + bytes.fromhex("08778F") + block_data[9:]
+                
+                # try writing with available keys
+                written = False
+                if keyB:
+                    try:
+                        self.cmd.mf1_write_one_block(4*s + b, MfcKeyType.B, keyB, block_data)
+                        written = True
+                    except Exception as e:
+                        pass
+                
+                if not written and keyA:
+                    try:
+                        self.cmd.mf1_write_one_block(4*s + b, MfcKeyType.A, keyA, block_data)
+                        written = True
+                    except Exception as e:
+                        pass
+                
+                if not written:
+                    print(f"Warning: Failed to write block {4*s + b}")
+                else:
+                    print(f"Wrote block {4*s + b}")
+        
+        print("Clone operation completed")
+
+
+
 @hf_mf.command('rdbl')
 class HFMFRDBL(MF1AuthArgsUnit):
     def args_parser(self) -> ArgumentParserNoExit:
@@ -1693,32 +1884,34 @@ def _run_mfkey32v2(items):
 
 
 class ItemGenerator:
-    def __init__(self, rs, i=0, j=1):
-        self.rs = rs
+    def __init__(self, rs, uid_found_keys = set()):
+        self.rs: list = rs
+        self.progress = 0
         self.i = 0
         self.j = 1
         self.found = set()
         self.keys = set()
+        for known_key in uid_found_keys:
+            self.test_key(known_key)
 
     def __iter__(self):
         return self
 
     def __next__(self):
-        try:
-            item_i = self.rs[self.i]
-        except IndexError:
-            raise StopIteration
-        if self.key_from_item(item_i) in self.found:
+        size = len(self.rs)
+        if self.j >= size:
             self.i += 1
+            if self.i >= size - 1:
+                raise StopIteration
             self.j = self.i + 1
-            return next(self)
-        try:
-            item_j = self.rs[self.j]
-        except IndexError:
+        item_i, item_j = self.rs[self.i], self.rs[self.j]
+        self.progress += 1
+        self.j += 1
+        if self.key_from_item(item_i) in self.found:
+            self.progress += max(0, size - self.j)
             self.i += 1
             self.j = self.i + 1
             return next(self)
-        self.j += 1
         if self.key_from_item(item_j) in self.found:
             return next(self)
         return item_i, item_j
@@ -1727,16 +1920,20 @@ def __next__(self):
     def key_from_item(item):
         return "{uid}-{nt}-{nr}-{ar}".format(**item)
 
-    def key_found(self, key, items):
-        self.keys.add(key)
-        for item in items:
-            try:
-                if item == self.rs[self.i]:
-                    self.i += 1
-                    self.j = self.i + 1
-            except IndexError:
-                break
-        self.found.update(self.key_from_item(item) for item in items)
+    def test_key(self, key, items = list()):
+        for item in self.rs:
+            item_key = self.key_from_item(item)
+            if item_key in self.found:
+                continue
+            if (item in items) or (Crypto1.mfkey32_is_reader_has_key(
+                int(item['uid'], 16),
+                int(item['nt'], 16),
+                int(item['nr'], 16),
+                int(item['ar'], 16),
+                key,
+            )):
+                self.keys.add(key)
+                self.found.add(item_key)
 
 
 @hf_mf.command('elog')
@@ -1749,7 +1946,7 @@ def args_parser(self) -> ArgumentParserNoExit:
         parser.add_argument('--decrypt', action='store_true', help="Decrypt key from MF1 log list")
         return parser
 
-    def decrypt_by_list(self, rs: list):
+    def decrypt_by_list(self, rs: list, uid_found_keys: set = set()):
         """
             Decrypt key from reconnaissance log list
 
@@ -1759,16 +1956,14 @@ def decrypt_by_list(self, rs: list):
         msg1 = f"  > {len(rs)} records => "
         msg2 = f"/{(len(rs)*(len(rs)-1))//2} combinations. "
         msg3 = " key(s) found"
-        n = 1
-        gen = ItemGenerator(rs)
+        gen = ItemGenerator(rs, uid_found_keys)
+        print(f"{msg1}{gen.progress}{msg2}{len(gen.keys)}{msg3}\r", end="")
         with Pool(cpu_count()) as pool:
             for result in pool.imap(_run_mfkey32v2, gen):
-                # TODO: if some keys already recovered, test them on item before running mfkey32 on item
                 if result is not None:
-                    gen.key_found(*result)
-                print(f"{msg1}{n}{msg2}{len(gen.keys)}{msg3}\r", end="")
-                n += 1
-        print()
+                    gen.test_key(*result)
+                print(f"{msg1}{gen.progress}{msg2}{len(gen.keys)}{msg3}\r", end="")
+        print(f"{msg1}{gen.progress}{msg2}{len(gen.keys)}{msg3}")
         return gen.keys
 
     def on_exec(self, args: argparse.Namespace):
@@ -1810,21 +2005,13 @@ def on_exec(self, args: argparse.Namespace):
             print(f" - Detection log for uid [{uid.upper()}]")
             result_maps_for_uid = result_maps[uid]
             for block in result_maps_for_uid:
-                print(f"  > Block {block} detect log decrypting...")
-                if 'A' in result_maps_for_uid[block]:
-                    # print(f" - A record: { result_maps[block]['A'] }")
-                    records = result_maps_for_uid[block]['A']
-                    if len(records) > 1:
-                        result_maps[uid][block]['A'] = self.decrypt_by_list(records)
-                    else:
-                        print(f"  > {len(records)} record")
-                if 'B' in result_maps_for_uid[block]:
-                    # print(f" - B record: { result_maps[block]['B'] }")
-                    records = result_maps_for_uid[block]['B']
-                    if len(records) > 1:
-                        result_maps[uid][block]['B'] = self.decrypt_by_list(records)
-                    else:
-                        print(f"  > {len(records)} record")
+                for keyType in 'AB':
+                    records = result_maps_for_uid[block][keyType] if keyType in result_maps_for_uid[block] else []
+                    if len(records) < 1:
+                        continue
+                    print(f"  > Decrypting block {block} key {keyType} detect log...")
+                    result_maps[uid][block][keyType] = self.decrypt_by_list(records, uid_found_keys)
+                    uid_found_keys.update(result_maps[uid][block][keyType])
             print("  > Result ---------------------------")
             for block in result_maps_for_uid.keys():
                 if 'A' in result_maps_for_uid[block]: