uraniborg: Support local cache pre-fetching during inclusion proof ch…

.gitignore

@@ -0,0 +1,9 @@
+# Python virtual environments and caches
+__pycache__
+*.pyc
+.pytest_cache
+.venv
+venv
+
+# macOS
+.DS_Store

uraniborg/.gitignore

@@ -11,12 +11,6 @@ gen/
 # Local config files (sdk path, etc)
 local.properties
 
-# Mac junks
-.DS_Store
-
-# Python related
-*.pyc
-
 # Some files specific to Android Studio
 *.iml
 .idea

uraniborg/README.md

@@ -28,6 +28,31 @@ Below are links to more specific documentations.
 ### Data Interpretation
 - [Interpreting Hubble results](docs/hubble_results.md)
 
+## Testing
+
+Unit tests for the Python automation and verification scripts are located in
+`scripts/python/tests/` and use the `pytest` framework:
+
+- `test_inclusion_proof_check.py`: Tests pre-fetching transparency log entries
+  (`--cache_prefetch_concurrency`, `--cache_prefetch_timeout`, `--cache_dir`),
+  opt-out (`--no_prefetch`), fail-open fallback on pre-fetch errors/timeouts,
+  input validation, exit codes (`0` when output is written vs. `1` on
+  execution/input error), and split inclusion verification.
+- `test_automate_observation.py`: Tests multi-device pre-fetch retry and latch
+  behavior across connected devices.
+
+To set up a virtual environment and run the test suite from the repository root:
+
+```bash
+# Set up a virtual environment and install pytest (one-time setup)
+python3 -m venv .venv
+source .venv/bin/activate
+pip install pytest
+
+# Run the test suite
+pytest uraniborg/scripts/python/tests/
+```
+
 ## Version
 The current version info can be found within the VERSION file, and in the
 build.gradle file of the Hubble app.

uraniborg/docs/automate_observation.md

@@ -26,3 +26,53 @@ INFO:automate_observation.py:main(561): Hubble output files can be found at: som
 If you see the final SUCCESS message, feel free to ignore earlier ERROR messages.
 Those resulted from some files that cannot be extracted from device, but does not
 affect the core files required for further analysis.
+
+## Performing Inclusion Proof Checks
+
+You can automatically verify extracted package APK splits against Android Binary
+Transparency logs by passing `--perform_inclusion_proof_check` along with the
+path to the `verifier` executable:
+
+```bash
+python3 automate_observation.py \
+  --perform_inclusion_proof_check \
+  --verifier_path=/path/to/verifier
+```
+
+The results are written to `packages_with_inclusion_proof_signal.txt` inside the
+device's results directory. Note that `inclusion_proof_check.py` (when invoked
+standalone or in CI) exits with code `0` whenever the check runs and writes the
+output JSON—even if individual APK splits fail their inclusion proof
+(`"inclusion_proof_verified": false`)—and exits with code `1` only when
+execution itself fails (e.g. missing/corrupt `packages.txt` or I/O errors).
+
+### Pre-fetching and Local Caching (Enabled by Default)
+When `--perform_inclusion_proof_check` is specified, `automate_observation.py`
+**automatically pre-fetches and caches transparency log entries locally** (using
+`verifier --fetch_entries`) before verifying individual packages. This avoids
+slow sequential HTTP downloads during per-package checks and reuses the local
+cache across multiple connected devices.
+
+You can customize caching and pre-fetching behavior with the following optional
+flags:
+* `--cache_prefetch_concurrency <N>`: Number of concurrent workers used when
+  pre-fetching Tessera entry tiles (default: `16`).
+* `--cache_prefetch_timeout <SECONDS>`: Timeout in seconds for the pre-fetching
+  step (default: `600`). If pre-fetching exceeds this ceiling (e.g. on a cold
+  cache over a slow or proxied link), it logs a warning and gracefully falls
+  back to on-demand tile fetching during verification. Because cached tile
+  writes are atomic, any tiles downloaded before the timeout are preserved in
+  the local cache and reused.
+* `--cache_dir <PATH>`: Custom root directory for the local cache (defaults to
+  the system user cache directory).
+* `--no_prefetch`: Disables pre-fetching log entries up front, falling back to
+  on-demand fetching during individual package verifications.
+
+### Running Unit Tests
+Unit tests for the inclusion proof check, pre-fetching, and multi-device
+workflows are located in `scripts/python/tests/`. You can run them with
+`pytest`:
+
+```bash
+pytest uraniborg/scripts/python/tests/
+```

uraniborg/scripts/python/automate_observation.py

@@ -78,10 +78,31 @@ def parse_arguments() -> argparse.Namespace:
                       help="If specified, after pulling results, perform "
                            "inclusion proof check for each package APK split "
                            "and write results to "
-                           "packages_with_inclusion_proof_signal.txt.")
+                           "packages_with_inclusion_proof_signal.txt. By "
+                           "default, transparency log entries are pre-fetched "
+                           "and cached locally beforehand.")
   parser.add_argument("--verifier_path", required=False,
                       help="Path to verifier executable, used if "
                            "--perform_inclusion_proof_check is specified.")
+  parser.add_argument("--cache_dir", required=False, default=None,
+                      help="Custom root directory for local cache used by "
+                           "verifier during inclusion proof checks. If "
+                           "unspecified, defaults to system cache directory.")
+  parser.add_argument("--cache_prefetch_concurrency", required=False, type=int,
+                      default=inclusion_proof_check.DEFAULT_PREFETCH_CONCURRENCY,
+                      help="Number of concurrent workers for fetching Tessera "
+                           "entry tiles when pre-fetching log entries during "
+                           "inclusion proof checks.")
+  parser.add_argument("--cache_prefetch_timeout", required=False, type=int,
+                      default=inclusion_proof_check.DEFAULT_PREFETCH_TIMEOUT,
+                      help="Timeout in seconds for pre-fetching transparency "
+                           "log entries before falling back to on-demand "
+                           "fetching.")
+  parser.add_argument("--no_prefetch", required=False, action="store_true",
+                      help="If specified, disables pre-fetching and caching of "
+                           "transparency log entries before running inclusion "
+                           "proof checks (pre-fetching is enabled by default "
+                           "when --perform_inclusion_proof_check is specified).")
   args = parser.parse_args()
 
   if args.perform_inclusion_proof_check and args.verifier_path is None:
@@ -807,6 +828,7 @@ def main():
     logger.warning("More than 1 device connected!")
 
   results = {}
+  prefetched = False
   for target_device in connected_devices:
     if target_device.unauthorized:
       logger.error("Please authorize device with serial number %s for ADB via "
@@ -864,8 +886,22 @@ def main():
 
     if args.perform_inclusion_proof_check:
       packages_txt_path = os.path.join(results_dir, "results", "packages.txt")
+      if (not args.no_prefetch and not prefetched and
+          os.path.isfile(packages_txt_path)):
+        prefetched = inclusion_proof_check.prefetch_log_entries(
+            args.verifier_path,
+            logger,
+            cache_dir=args.cache_dir,
+            concurrency=args.cache_prefetch_concurrency,
+            timeout=args.cache_prefetch_timeout)
       inclusion_proof_check.perform_inclusion_proof_check(
-          args.verifier_path, packages_txt_path, logger)
+          args.verifier_path,
+          packages_txt_path,
+          logger,
+          cache_dir=args.cache_dir,
+          concurrency=args.cache_prefetch_concurrency,
+          timeout=args.cache_prefetch_timeout,
+          prefetch=False)
 
   for device in results:
     logger.info("SUCCESS! Hubble was successfully deployed and executed on "

uraniborg/scripts/python/inclusion_proof_check.py

@@ -21,16 +21,78 @@
 import logging
 import os
 import subprocess
+import sys
 import tempfile
+from typing import Optional
 
 OUTPUT_FILENAME = 'packages_with_inclusion_proof_signal.txt'
+DEFAULT_PREFETCH_CONCURRENCY = 16
+DEFAULT_PREFETCH_TIMEOUT = 600
+
+
+def prefetch_log_entries(verifier_executable: str,
+                         logger: logging.Logger,
+                         cache_dir: Optional[str] = None,
+                         concurrency: int = DEFAULT_PREFETCH_CONCURRENCY,
+                         timeout: int = DEFAULT_PREFETCH_TIMEOUT) -> bool:
+  """Pre-fetches and locally caches transparency log entries up to checkpoint.
+
+  Args:
+    verifier_executable: path to verifier tool.
+    logger: logger instance.
+    cache_dir: optional custom root directory for local cache.
+    concurrency: number of concurrent worker threads for fetching Tessera tiles.
+    timeout: maximum time in seconds to wait for pre-fetching before timing out.
+
+  Returns:
+    True if pre-fetching succeeded, False otherwise.
+  """
+  try:
+    cmd = [
+        verifier_executable,
+        "--log_type=google_1p_apk",
+        "--fetch_entries",
+        f"--concurrency={concurrency}",
+    ]
+    if cache_dir:
+      cmd.append(f"--cache_dir={cache_dir}")
+
+    logger.info("Pre-fetching transparency log entries (concurrency=%d)...",
+                concurrency)
+    logger.debug("Running verifier prefetch: %s", " ".join(cmd))
+    result = subprocess.run(cmd, check=False, timeout=timeout)
+    if result.returncode == 0:
+      logger.info("Successfully pre-fetched and cached log entries.")
+      return True
+    else:
+      logger.warning(
+          "Pre-fetching log entries exited with code %d. "
+          "Falling back to on-demand tile fetching during verification.",
+          result.returncode)
+      return False
+  except subprocess.TimeoutExpired:
+    logger.warning(
+        "Pre-fetching log entries timed out after %d seconds. "
+        "Falling back to on-demand tile fetching during verification.",
+        timeout)
+    return False
+  except FileNotFoundError:
+    logger.error("`%s` command not found.", verifier_executable)
+    return False
+  except Exception as e:
+    logger.warning("Error pre-fetching log entries: %s. Continuing...", e)
+    return False
+
 
 def run_verifier(verifier_executable: str, payload_path: str,
-                 logger: logging.Logger) -> bool:
+                 logger: logging.Logger,
+                 cache_dir: Optional[str] = None) -> bool:
   """Runs verifier tool and returns True if inclusion proof is successful."""
   try:
     cmd = [verifier_executable, f"--payload_path={payload_path}",
            "--log_type=google_1p_apk"]
+    if cache_dir:
+      cmd.append(f"--cache_dir={cache_dir}")
     with open(payload_path, "r") as f_in:
       payload = f_in.read()
       logger.debug("payload content: %s", payload)
@@ -53,33 +115,56 @@ def run_verifier(verifier_executable: str, payload_path: str,
     return False
 
 
-def perform_inclusion_proof_check(verifier_executable: str,
-                                  packages_file_path: str,
-                                  logger: logging.Logger):
+def perform_inclusion_proof_check(
+    verifier_executable: str,
+    packages_file_path: str,
+    logger: logging.Logger,
+    cache_dir: Optional[str] = None,
+    concurrency: int = DEFAULT_PREFETCH_CONCURRENCY,
+    timeout: int = DEFAULT_PREFETCH_TIMEOUT,
+    prefetch: bool = True) -> bool:
   """Reads packages.txt and performs inclusion proof check for each APK split.
 
-  It writes results to file which name defined in OUTPUT_FILENAME in same dir.
+  By default, pre-fetches and locally caches transparency log entries before
+  verifying individual package splits. Writes results to file defined in
+  OUTPUT_FILENAME in the same directory as packages.txt.
 
   Args:
     verifier_executable: path to verifier tool.
     packages_file_path: path to packages.txt.
     logger: logger instance.
+    cache_dir: optional custom root directory for local cache.
+    concurrency: number of concurrent workers for fetching Tessera entry tiles.
+    timeout: maximum time in seconds to wait for pre-fetching before timing out.
+    prefetch: whether to pre-fetch log entries before verifying packages.
+
+  Returns:
+    True if packages.txt was valid and inclusion proof results were successfully
+    written to disk, False otherwise.
   """
   if not os.path.isfile(packages_file_path):
     logger.error("packages.txt not found at %s", packages_file_path)
-    return
+    return False
 
   logger.info("Performing inclusion proof check...")
   try:
     with open(packages_file_path, "r") as f_in:
       packages_json = json.load(f_in)
   except json.JSONDecodeError as e:
     logger.error("Failed to parse %s: %s", packages_file_path, e)
-    return
+    return False
 
-  if "packages" not in packages_json:
-    logger.error("No 'packages' key in %s", packages_file_path)
-    return
+  if not isinstance(packages_json.get("packages"), list):
+    logger.error("No valid 'packages' list found in %s", packages_file_path)
+    return False
+
+  if prefetch and packages_json["packages"]:
+    prefetch_log_entries(
+        verifier_executable,
+        logger,
+        cache_dir=cache_dir,
+        concurrency=concurrency,
+        timeout=timeout)
 
   for package in packages_json["packages"]:
     if "name" not in package or "versionCode" not in package:
@@ -121,7 +206,11 @@ def perform_inclusion_proof_check(verifier_executable: str,
           fp.write(payload)
           temp_payload_path = fp.name
 
-        verified = run_verifier(verifier_executable, temp_payload_path, logger)
+        verified = run_verifier(
+            verifier_executable,
+            temp_payload_path,
+            logger,
+            cache_dir=cache_dir)
         split["inclusion_proof_verified"] = verified
       finally:
         if temp_payload_path and os.path.exists(temp_payload_path):
@@ -149,18 +238,41 @@ def perform_inclusion_proof_check(verifier_executable: str,
     with open(output_path, "w") as f_out:
       json.dump(output_json, f_out, indent=2)
     logger.info("Inclusion proof results written to %s", output_path)
+    return True
   except Exception as e:
     logger.error("Failed to write results to %s: %s", output_path, e)
+    return False
 
 
 def main():
   parser = argparse.ArgumentParser(
-      description="Perform inclusion proof check on packages.txt.",
+      description="Perform inclusion proof check on packages.txt. By default, "
+                  "pre-fetches and caches transparency log entries locally "
+                  "before verifying individual package splits. Exits 0 when "
+                  "results are written to disk (even if individual splits fail "
+                  "verification) and exits 1 only on input or execution errors.",
       formatter_class=argparse.ArgumentDefaultsHelpFormatter)
   parser.add_argument("--packages_file", required=True,
                       help="Path to packages.txt file.")
   parser.add_argument("--verifier_path", required=True,
                       help="Path to verifier executable.")
+  parser.add_argument("--cache_dir", required=False, default=None,
+                      help="Custom root directory for local cache used by "
+                           "verifier. If unspecified, defaults to system cache "
+                           "directory.")
+  parser.add_argument("--cache_prefetch_concurrency", required=False, type=int,
+                      default=DEFAULT_PREFETCH_CONCURRENCY,
+                      help="Number of concurrent workers for fetching Tessera "
+                           "entry tiles when pre-fetching log entries.")
+  parser.add_argument("--cache_prefetch_timeout", required=False, type=int,
+                      default=DEFAULT_PREFETCH_TIMEOUT,
+                      help="Timeout in seconds for pre-fetching transparency "
+                           "log entries before falling back to on-demand "
+                           "fetching.")
+  parser.add_argument("--no_prefetch", required=False, action="store_true",
+                      help="If specified, disables pre-fetching and caching of "
+                           "transparency log entries before running inclusion "
+                           "proof checks.")
   parser.add_argument("-D", "--debug", required=False, action="store_true",
                       help="If specified, debugging mode is turned on.")
   args = parser.parse_args()
@@ -176,7 +288,15 @@ def main():
   s_handler.setFormatter(s_format)
   logger.addHandler(s_handler)
 
-  perform_inclusion_proof_check(args.verifier_path, args.packages_file, logger)
+  if not perform_inclusion_proof_check(
+      args.verifier_path,
+      args.packages_file,
+      logger,
+      cache_dir=args.cache_dir,
+      concurrency=args.cache_prefetch_concurrency,
+      timeout=args.cache_prefetch_timeout,
+      prefetch=not args.no_prefetch):
+    sys.exit(1)
 
 
 if __name__ == "__main__":