and script to browse crashes

crash_reporting_cloud_function/log_browser.py

@@ -0,0 +1,51 @@
+import tarfile
+from io import BytesIO
+from pathlib import Path
+
+from google.cloud import storage
+
+CRASH_REPORT_BUCKET = "crash-reports"
+GCP_PROJECT = "decoded-cove-239422"
+OUTPUT = Path("crashes")
+
+if OUTPUT.exists() is False:
+    OUTPUT.mkdir()
+
+
+def process_crash_report(crash_report_contents: str) -> None:
+    crash_report_lines = crash_report_contents.splitlines()
+    tweaks = []
+    last_exc_bt = None
+    dylib_occur = 0
+    for idx, crash_line in enumerate(crash_report_lines):
+        if "Last Exception Backtrace:" in crash_line:
+            last_exc_bt = crash_report_lines[idx + 1]
+        if "/Library/MobileSubstrate/DynamicLibraries/" in crash_line:
+            tweaks.append(crash_line)
+        if " carplayenable.dylib " in crash_line:
+            dylib_occur += 1
+
+    num_tweaks = len(tweaks)
+    if num_tweaks > 40 or last_exc_bt is None or dylib_occur < 3:
+        return 0
+    print(f"tweaks: {num_tweaks} dylib occur {dylib_occur}\n{last_exc_bt}\n")
+    for tweak in tweaks:
+        print(tweak)
+    print("\n\n")
+    return 1
+
+
+storage_client = storage.Client(project=GCP_PROJECT)
+all_blobs = storage_client.list_blobs(CRASH_REPORT_BUCKET)
+for blob in all_blobs:
+
+    crash_tar_bytes = blob.download_as_bytes()
+    with tarfile.open(fileobj=BytesIO(crash_tar_bytes), mode="r:*") as tarf:
+        for member in tarf.getmembers():
+            if member.isdir():
+                continue
+            file_contents = tarf.extractfile(member).read()
+            if process_crash_report(file_contents.decode("utf-8")) == 1:
+                shortname = Path(member.name).name
+                with open(OUTPUT / shortname, mode="wb") as f:
+                    f.write(file_contents)

crash_reporting_cloud_function/main.py

@@ -7,6 +7,7 @@
 CRASH_REPORT_BUCKET = "crash-reports"
 GCP_PROJECT = "decoded-cove-239422"
 
+
 def find_value_for_key(key: bytes, contents: bytes) -> str:
     for content_line in contents.split(b"\n"):
         if key in content_line:

crash_reporting_cloud_function/tasks.py

@@ -1,5 +1,8 @@
 from invoke import task
 
+
 @task
 def deploy(ctx):
-    ctx.run("gcloud functions deploy process_crash_reports --runtime python37 --project decoded-cove-239422 --trigger-http --allow-unauthenticated")
+    ctx.run(
+        "gcloud functions deploy process_crash_reports --runtime python37 --project decoded-cove-239422 --trigger-http --allow-unauthenticated"
+    )

repo/push_deb_to_repo.py

@@ -16,12 +16,14 @@ def run_local_command(command: str) -> Optional[str]:
     output = subprocess.check_output(command, shell=True)
     return output.decode("utf-8") if output else None
 
+
 def run_remote_command(command: str) -> Optional[str]:
-    full_command = f"ssh root@{REPO_IP_ADDRESS} \"{command}\""
+    full_command = f'ssh root@{REPO_IP_ADDRESS} "{command}"'
     return run_local_command(full_command)
 
+
 def send_file(local_file: str, remote_path: str) -> None:
-    run_local_command(f"scp \"{local_file}\" root@{REPO_IP_ADDRESS}:\"{REPO_PATH}/{remote_path}\"")
+    run_local_command(f'scp "{local_file}" root@{REPO_IP_ADDRESS}:"{REPO_PATH}/{remote_path}"')
 
 
 def run(deb_to_upload: Path) -> None:
@@ -60,7 +62,9 @@ def run(deb_to_upload: Path) -> None:
     package_info += f"Filename: {remote_deb_location}\n"
 
     # Gzip package file and upload
-    with tempfile.NamedTemporaryFile(mode="wb", delete=False) as packagef, gzip.GzipFile(fileobj=packagef, mode="wb") as gzout:
+    with tempfile.NamedTemporaryFile(mode="wb", delete=False) as packagef, gzip.GzipFile(
+        fileobj=packagef, mode="wb"
+    ) as gzout:
         # Write contents to temp file
         gzout.write(package_info.encode("utf-8"))
         gzout.flush()