Add uv github action and handle stderr logging.

.github/workflows/update-uv.yaml

@@ -0,0 +1,35 @@
+name: update-uv
+
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  lock:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: astral-sh/setup-uv@v3
+
+      - run: |
+          echo "\`\`\`" > uv_output.md
+          uv lock --upgrade 2>&1 | tee -a uv_output.md
+          echo "\`\`\`" >> uv_output.md
+
+      - name: Create pull request
+        uses: peter-evans/create-pull-request@v7
+        with:
+          committer: hacktobeer <ramsesdebeer@gmail.com>
+          token: ${{ secrets.GITHUB_TOKEN }}
+          commit-message: Update uv lockfile
+          title: Update uv lockfile
+          body-path: uv_output.md
+          branch: update-uv
+          base: main
+          labels: install
+          delete-branch: true
+          add-paths: uv.lock

src/image_export.py

@@ -28,6 +28,7 @@
 from openrelik_worker_common.task_utils import create_task_result, get_input_files
 
 from .app import celery
+from .utils import process_plaso_cli_logs
 
 
 # These are taken from Plaso's extraction tool.
@@ -222,15 +223,15 @@ def _get_base_command(export_directory):
             logger.info(f"Executing command: {' '.join(command)}")
             # Execute the command and block until it finishes.
             process = subprocess.Popen(
-                command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+                command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1
             )
             while process.poll() is None:
                 self.send_event("task-progress")
                 time.sleep(1)
             if process.stdout:
                 logger.info(process.stdout.read())
             if process.stderr:
-                logger.error(process.stderr.read())
+                process_plaso_cli_logs(process.stderr.read(), logger)
 
         for export_directory in export_directories:
             export_directory_path = Path(export_directory)

src/utils.py

@@ -0,0 +1,55 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+import re
+
+def process_plaso_cli_logs(cli_logs: str, logger) -> None:
+    """Process Plaso CLI log output to relevant Python log sink and level.
+
+    Args:
+        cli_logs: The input logs from the plaso CLI tool.
+        logger: The logger to use for logging.
+
+    Returns:
+        None
+    """
+    # Matches "[LEVEL] Message" as defined in the log2timeline source code as
+    # "format='[%(levelname)s] %(message)s')""
+    log_pattern = re.compile(r"^\[(?P<level>\w+)\]\s*(?P<msg>.*)$")
+
+    # Default state
+    current_level = logging.INFO
+
+    for line in cli_logs.splitlines():
+        line = line.rstrip()
+        if not line:
+            continue
+
+        match = log_pattern.match(line)
+
+        if match:
+            # We found a new header! Update the level.
+            # We need this to handle multi-line logs like tracebacks.
+            level_str = match.group("level").upper()
+            message = match.group("msg")
+
+            # Use INFO as a fallback if getLevelName returns a string (meaning not found)
+            new_level = logging.getLevelName(level_str)
+            if isinstance(new_level, int):
+                current_level = new_level
+
+            logger.log(current_level, f"{message}")
+        else:
+            # This line doesn't have a [LEVEL] prefix.
+            logger.log(current_level, f"{line}")