Merge pull request #14 from LedgerHQ/tdj/accept_LL_logs_as_input

allow ledger live log directly as dale input and manage the log inversion

src/dale/logger.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 
+import json
 import logging
 from argparse import ArgumentParser, RawTextHelpFormatter
 from pathlib import Path
@@ -13,24 +14,49 @@
 def init_parser() -> ArgumentParser:
     parser = ArgumentParser(description="Explicit logging of a list of APDUs",
                             formatter_class=RawTextHelpFormatter)
-    parser.add_argument("apdu_file", metavar="APDU_FILE", help="The file containing the list of APDUs", type=Path)
+    parser.add_argument("apdu_file", metavar="APDU_FILE",
+                        help="The file containing the list of APDUs",
+                        type=Path)
+    parser.add_argument("--json_input", action="store_true",
+                        help="Read a Ledger Live log instead of a raw APDU file")
+    parser.add_argument("--reverse", action="store_true",
+                        help="Read APDU file in reverse order (bottom to top)")
     return parser
 
 
 def main():
     logging.root.setLevel(logging.INFO)
 
+    apdus: list[str] = []
     parser = init_parser()
     args = parser.parse_args()
     apdu_file = args.apdu_file.resolve()
     if not apdu_file.is_file():
         raise AssertionError(f"'{apdu_file}' does not exist or is not a file! Aborting")
     logging.info("Reading from %s", apdu_file)
 
-    with apdu_file.open() as filee:
-        with DefaultAPDUParser([ExchangeFactory(), DefaultAPDUsFactory(), Factory()]) as apdu_parser:
-            for line in filee:
-                apdu_parser.feed(line)
+    # check if input file is Ledger Live log
+    if args.json_input:
+        logging.info("Parsing Ledger Live log file")
+        with apdu_file.open("r", encoding="utf8") as f:
+            entries = json.load(f)
+
+            # Extract the APDUs from the log
+            # get the "message" field from each entry where the "type" is "apdu"
+            apdus = list(map(lambda x: x["message"], filter(lambda x: x.get("type", "") == "apdu", entries)))
+    else:
+        logging.info("Reading raw APDU file")
+        with apdu_file.open() as file:
+            for line in file:
+                apdus.append(line)
+
+    if args.reverse:
+        logging.info("Reversing lines")
+        apdus.reverse()
+
+    with DefaultAPDUParser([ExchangeFactory(), DefaultAPDUsFactory(), Factory()]) as apdu_parser:
+        for apdu in apdus:
+            apdu_parser.feed(apdu)
 
     for exchange in apdu_parser.conversation:
         print(str(exchange))