Add some fixes for security vulnerabilities

src/amd_debug/common.py

@@ -13,6 +13,8 @@
 import struct
 import subprocess
 import sys
+import stat
+import pathlib
 from ast import literal_eval
 from datetime import date, timedelta
 
@@ -133,18 +135,46 @@ def _configure_log(prefix) -> str:
         path = os.environ.get("XDG_DATA_HOME") or os.path.join(
             home, ".local", "share", "amd-debug-tools"
         )
-        os.makedirs(path, exist_ok=True)
-        log = os.path.join(
-            path,
-            f"{prefix}-{date.today()}.txt",
-        )
-        if not os.path.exists(log):
-            with open(log, "w", encoding="utf-8") as f:
-                f.write("")
+
+        # Prevent symlink attacks on log directory/file when running as root
+        path_obj = pathlib.Path(path)
+
+        try:
+            if path_obj.exists():
+                if path_obj.is_symlink():
+                    raise ValueError(f"Log directory is a symlink: {path}")
+            else:
+                os.makedirs(path, mode=0o755, exist_ok=True)
+                if path_obj.is_symlink():
+                    raise ValueError(f"Log directory became a symlink during creation: {path}")
+
             if "SUDO_UID" in os.environ:
-                os.chown(path, int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"]))
-                os.chown(log, int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"]))
-        level = logging.DEBUG
+                os.lchown(path, int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"]))
+        except (ValueError, OSError) as e:
+            logging.warning(f"Cannot create safe log directory: {e}")
+            log = "/dev/null"
+            level = logging.WARNING
+        else:
+            log = os.path.join(path, f"{prefix}-{date.today()}.txt")
+
+            try:
+                fd = os.open(
+                    log,
+                    os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW,
+                    mode=stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH,  # 0o644
+                )
+                try:
+                    os.close(fd)
+                    if "SUDO_UID" in os.environ:
+                        os.lchown(log, int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"]))
+                except Exception:
+                    os.close(fd)
+                    raise
+            except FileExistsError:
+                if os.path.islink(log):
+                    raise ValueError(f"Log file is a symlink: {log}")
+
+            level = logging.DEBUG
     else:
         log = "/dev/null"
         level = logging.WARNING

src/amd_debug/sleep_report.py

@@ -3,10 +3,14 @@
 import os
 import re
 import math
+import stat
+import pathlib
+import html
 from datetime import datetime, timedelta
 import numpy as np
 from tabulate import tabulate
 from jinja2 import Environment, FileSystemLoader
+from markupsafe import Markup
 import pandas as pd
 
 from amd_debug.database import SleepDatabase
@@ -288,7 +292,9 @@ def get_prereq_data(self):
                 if self.format == "html" and [
                     table for table in tables if table in content
                 ]:
-                    content = self.convert_table_dataframe(content)
+                    content = Markup(self.convert_table_dataframe(content))
+                elif self.format == "html":
+                    content = Markup(html.escape(content))
                 prereq_debug.append({"data": f"{content.strip()}"})
         return prereq, t0, prereq_debug
 
@@ -342,8 +348,8 @@ def get_cycle_data(self):
             if self.format == "html":
                 data = ""
                 for line in self.db.report_cycle_data(cycle).split("\n"):
-                    data += f"<p>{line}</p>"
-                cycles.append({"cycle_num": num, "data": data})
+                    data += f"<p>{html.escape(line)}</p>"
+                cycles.append({"cycle_num": num, "data": Markup(data)})
             else:
                 cycles.append([num, self.db.report_cycle_data(cycle)])
             if self.debug:
@@ -354,18 +360,21 @@ def get_cycle_data(self):
                     if self.format == "html" and [
                         table for table in tables if table in content
                     ]:
-                        content = self.convert_table_dataframe(content)
+                        content = Markup(self.convert_table_dataframe(content))
+                    elif self.format == "html":
+                        content = Markup(html.escape(content))
                     messages.append(content)
                     priorities.append(get_log_priority(row[1]))
 
-                # Add formatted power rail summary
                 cycle_row = self.df[self.df["Start Time"] == cycle]
                 if not cycle_row.empty:
                     duration = cycle_row["Duration"].iloc[0]
                     power_rail_summary = self.format_power_rail_data(cycle, duration)
                     if power_rail_summary:
+                        if self.format == "html":
+                            power_rail_summary = Markup(html.escape(power_rail_summary))
                         messages.append(power_rail_summary)
-                        priorities.append(get_log_priority(6))  # Info level
+                        priorities.append(get_log_priority(6))
 
                 debug.append(
                     {"cycle_num": num, "messages": messages, "priorities": priorities}
@@ -379,7 +388,10 @@ def build_template(self, inc_prereq) -> str:
 
         # Load the template
         p = os.path.dirname(amd_debug.__file__)
-        environment = Environment(loader=FileSystemLoader(os.path.join(p, "templates")))
+        environment = Environment(
+            loader=FileSystemLoader(os.path.join(p, "templates")),
+            autoescape=True
+        )
         template = environment.get_template(self.format)
 
         # Load the prereq data
@@ -469,12 +481,36 @@ def build_template(self, inc_prereq) -> str:
             "failures": failures,
         }
         if self.fname:
-            with open(self.fname, "w", encoding="utf-8") as f:
-                f.write(template.render(context))
-            if "SUDO_UID" in os.environ:
-                os.chown(
-                    self.fname, int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"])
+            try:
+                resolved = pathlib.Path(self.fname).resolve()
+            except (ValueError, OSError) as e:
+                raise ValueError(f"Invalid report file path: {self.fname}") from e
+
+            # Prevent symlink attack when running as root
+            try:
+                fd = os.open(
+                    self.fname,
+                    os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+                    mode=stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH,  # 0o644
                 )
+            except FileExistsError:
+                raise FileExistsError(
+                    f"Report file already exists: {self.fname}. "
+                    "Please remove it or specify a different filename."
+                ) from None
+
+            try:
+                with os.fdopen(fd, "w", encoding="utf-8") as f:
+                    f.write(template.render(context))
+                if "SUDO_UID" in os.environ:
+                    os.fchown(fd, int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"]))
+            except Exception:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+                raise
+
             return "Report written to {f}".format(f=self.fname)
         else:
             return template.render(context)