Merge branch 'master' of https://github.com/sunnypilot/sunnypilot into sync

.github/workflows/tests.yaml

@@ -107,8 +107,8 @@ jobs:
 
   build_mac:
     name: build macOS
-    if: false # tmp disable due to brew install not working
     runs-on: ${{ ((github.repository == 'commaai/openpilot') && ((github.event_name != 'pull_request') || (github.event.pull_request.head.repo.full_name == 'commaai/openpilot'))) && 'namespace-profile-macos-8x14' || 'macos-latest' }}
+    if: false  # There'll be one day that this works. That day is not today.
     steps:
     - uses: actions/checkout@v4
       with:

selfdrive/ui/mici/onroad/confidence_ball.py

@@ -6,6 +6,8 @@ from openpilot.system.ui.widgets import Widget
 from openpilot.system.ui.lib.application import gui_app
 from openpilot.common.filter_simple import FirstOrderFilter
 
+from openpilot.selfdrive.ui.sunnypilot.mici.onroad.confidence_ball import ConfidenceBallSP
+
 
 def draw_circle_gradient(center_x: float, center_y: float, radius: int,
                          top: rl.Color, bottom: rl.Color) -> None:
@@ -21,9 +23,10 @@ def draw_circle_gradient(center_x: float, center_y: float, radius: int,
                20, rl.BLACK)
 
 
-class ConfidenceBall(Widget):
+class ConfidenceBall(Widget, ConfidenceBallSP):
   def __init__(self, demo: bool = False):
-    super().__init__()
+    Widget.__init__(self)
+    ConfidenceBallSP.__init__(self)
     self._demo = demo
     self._confidence_filter = FirstOrderFilter(-0.5, 0.5, 1 / gui_app.target_fps)
 
@@ -37,6 +40,8 @@ class ConfidenceBall(Widget):
     # animate status dot in from bottom
     if ui_state.status == UIStatus.DISENGAGED:
       self._confidence_filter.update(-0.5)
+    elif ui_state.status in (UIStatus.LAT_ONLY, UIStatus.LONG_ONLY):
+      self._confidence_filter.update(1 - max(self.get_animate_status_probs() or [1]))
     else:
       self._confidence_filter.update((1 - max(ui_state.sm['modelV2'].meta.disengagePredictions.brakeDisengageProbs or [1])) *
                                                         (1 - max(ui_state.sm['modelV2'].meta.disengagePredictions.steerOverrideProbs or [1])))
@@ -65,6 +70,9 @@ class ConfidenceBall(Widget):
         top_dot_color = rl.Color(255, 0, 21, 255)
         bottom_dot_color = rl.Color(255, 0, 89, 255)
 
+    elif ui_state.status in (UIStatus.LAT_ONLY, UIStatus.LONG_ONLY):
+      top_dot_color = bottom_dot_color = self.get_lat_long_dot_color()
+
     elif ui_state.status == UIStatus.OVERRIDE:
       top_dot_color = rl.Color(255, 255, 255, 255)
       bottom_dot_color = rl.Color(82, 82, 82, 255)

selfdrive/ui/mici/onroad/model_renderer.py

@@ -12,6 +12,8 @@ from openpilot.system.ui.lib.application import gui_app
 from openpilot.system.ui.lib.shader_polygon import draw_polygon, Gradient
 from openpilot.system.ui.widgets import Widget
 
+from openpilot.selfdrive.ui.sunnypilot.mici.onroad.model_renderer import LANE_LINE_COLORS_SP
+
 CLIP_MARGIN = 500
 MIN_DRAW_DISTANCE = 10.0
 MAX_DRAW_DISTANCE = 100.0
@@ -32,6 +34,7 @@ LANE_LINE_COLORS = {
   UIStatus.DISENGAGED: rl.Color(200, 200, 200, 255),
   UIStatus.OVERRIDE: rl.Color(255, 255, 255, 255),
   UIStatus.ENGAGED: rl.Color(0, 255, 64, 255),
+  **LANE_LINE_COLORS_SP,
 }
 
 

selfdrive/ui/mici/onroad/torque_bar.py

@@ -185,13 +185,13 @@ class TorqueBar(Widget):
 
     # animate alpha and angle span
     if not self._demo:
-      self._torque_line_alpha_filter.update(ui_state.status != UIStatus.DISENGAGED)
+      self._torque_line_alpha_filter.update(ui_state.status not in (UIStatus.DISENGAGED, UIStatus.LONG_ONLY))
     else:
       self._torque_line_alpha_filter.update(1.0)
 
     torque_line_bg_alpha = np.interp(abs(self._torque_filter.x), [0.5, 1.0], [0.25, 0.5])
     torque_line_bg_color = rl.Color(255, 255, 255, int(255 * torque_line_bg_alpha * self._torque_line_alpha_filter.x))
-    if ui_state.status != UIStatus.ENGAGED and not self._demo:
+    if ui_state.status not in (UIStatus.ENGAGED, UIStatus.LAT_ONLY) and not self._demo:
       torque_line_bg_color = rl.Color(255, 255, 255, int(255 * 0.15 * self._torque_line_alpha_filter.x))
 
     # draw curved line polygon torque bar
@@ -234,7 +234,7 @@ class TorqueBar(Widget):
       max(0, abs(self._torque_filter.x) - 0.75) * 4,
     )
 
-    if ui_state.status != UIStatus.ENGAGED and not self._demo:
+    if ui_state.status not in (UIStatus.ENGAGED, UIStatus.LAT_ONLY) and not self._demo:
       start_color = end_color = rl.Color(255, 255, 255, int(255 * 0.35 * self._torque_line_alpha_filter.x))
 
     gradient = Gradient(

selfdrive/ui/onroad/augmented_road_view.py

@@ -17,6 +17,8 @@ from openpilot.common.transformations.orientation import rot_from_euler
 if gui_app.sunnypilot_ui():
   from openpilot.selfdrive.ui.sunnypilot.onroad.hud_renderer import HudRendererSP as HudRenderer
 
+from openpilot.selfdrive.ui.sunnypilot.onroad.augmented_road_view import BORDER_COLORS_SP
+
 OpState = log.SelfdriveState.OpenpilotState
 CALIBRATED = log.LiveCalibrationData.Status.calibrated
 ROAD_CAM = VisionStreamType.VISION_STREAM_ROAD
@@ -27,6 +29,7 @@ BORDER_COLORS = {
   UIStatus.DISENGAGED: rl.Color(0x12, 0x28, 0x39, 0xFF),  # Blue for disengaged state
   UIStatus.OVERRIDE: rl.Color(0x89, 0x92, 0x8D, 0xFF),  # Gray for override state
   UIStatus.ENGAGED: rl.Color(0x16, 0x7F, 0x40, 0xFF),  # Green for engaged state
+  **BORDER_COLORS_SP,
 }
 
 WIDE_CAM_MAX_SPEED = 10.0  # m/s (22 mph)

selfdrive/ui/onroad/model_renderer.py

@@ -11,6 +11,8 @@ from openpilot.system.ui.lib.application import gui_app
 from openpilot.system.ui.lib.shader_polygon import draw_polygon, Gradient
 from openpilot.system.ui.widgets import Widget
 
+from openpilot.selfdrive.ui.sunnypilot.onroad.model_renderer import ChevronMetrics, ModelRendererSP
+
 CLIP_MARGIN = 500
 MIN_DRAW_DISTANCE = 10.0
 MAX_DRAW_DISTANCE = 100.0
@@ -41,9 +43,11 @@ class LeadVehicle:
   fill_alpha: int = 0
 
 
-class ModelRenderer(Widget):
+class ModelRenderer(Widget, ChevronMetrics, ModelRendererSP):
   def __init__(self):
-    super().__init__()
+    Widget.__init__(self)
+    ChevronMetrics.__init__(self)
+    ModelRendererSP.__init__(self)
     self._longitudinal_control = False
     self._experimental_mode = False
     self._blend_filter = FirstOrderFilter(1.0, 0.25, 1 / gui_app.target_fps)
@@ -128,6 +132,7 @@ class ModelRenderer(Widget):
 
     if render_lead_indicator and radar_state:
       self._draw_lead_indicator()
+      self.chevron_metrics.draw_lead_status(sm, radar_state, self._rect, self._lead_vehicles)
 
   def _update_raw_points(self, model):
     """Update raw 3D points from model data"""
@@ -281,6 +286,10 @@ class ModelRenderer(Widget):
     allow_throttle = sm['longitudinalPlan'].allowThrottle or not self._longitudinal_control
     self._blend_filter.update(int(allow_throttle))
 
+    if ui_state.rainbow_path:
+      self.rainbow_path.draw_rainbow_path(self._rect, self._path)
+      return
+
     if self._experimental_mode:
       # Draw with acceleration coloring
       if len(self._exp_gradient.colors) > 1:

selfdrive/ui/sunnypilot/mici/onroad/confidence_ball.py

@@ -0,0 +1,26 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+from openpilot.selfdrive.ui.onroad.augmented_road_view import BORDER_COLORS
+from openpilot.selfdrive.ui.ui_state import ui_state, UIStatus
+
+
+class ConfidenceBallSP:
+  @staticmethod
+  def get_animate_status_probs():
+    if ui_state.status == UIStatus.LAT_ONLY:
+      return ui_state.sm['modelV2'].meta.disengagePredictions.steerOverrideProbs
+
+    # UIStatus.LONG_ONLY
+    return ui_state.sm['modelV2'].meta.disengagePredictions.brakeDisengageProbs
+
+  @staticmethod
+  def get_lat_long_dot_color():
+    if ui_state.status == UIStatus.LAT_ONLY:
+      return BORDER_COLORS[UIStatus.LAT_ONLY]
+
+    # UIStatus.LONG_ONLY
+    return BORDER_COLORS[UIStatus.LONG_ONLY]

selfdrive/ui/sunnypilot/mici/onroad/model_renderer.py

@@ -0,0 +1,13 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+import pyray as rl
+from openpilot.selfdrive.ui.ui_state import UIStatus
+
+LANE_LINE_COLORS_SP = {
+  UIStatus.LAT_ONLY: rl.Color(0, 255, 64, 255),
+  UIStatus.LONG_ONLY: rl.Color(0, 255, 64, 255),
+}

selfdrive/ui/sunnypilot/onroad/augmented_road_view.py

@@ -0,0 +1,13 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+import pyray as rl
+from openpilot.selfdrive.ui.ui_state import UIStatus
+
+BORDER_COLORS_SP = {
+  UIStatus.LAT_ONLY: rl.Color(0x00, 0xC8, 0xC8, 0xFF),  # Cyan for lateral-only state
+  UIStatus.LONG_ONLY: rl.Color(0x96, 0x1C, 0xA8, 0xFF),  # Purple for longitudinal-only state
+}

selfdrive/ui/sunnypilot/onroad/chevron_metrics.py

@@ -0,0 +1,147 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+import numpy as np
+
+import pyray as rl
+from openpilot.common.constants import CV
+from openpilot.selfdrive.ui.ui_state import ui_state
+from openpilot.system.ui.lib.application import gui_app, FontWeight
+from openpilot.system.ui.lib.text_measure import measure_text_cached
+
+
+class ChevronOptions:
+  OFF = 0
+  DISTANCE_ONLY = 1
+  SPEED_ONLY = 2
+  TTC_ONLY = 3
+  ALL = 4
+
+
+class ChevronMetrics:
+  def __init__(self):
+    self._lead_status_alpha: float = 0.0
+    self._font = gui_app.font(FontWeight.SEMI_BOLD)
+
+  def update_alpha(self, has_lead: bool):
+    """Update the alpha value for fade in/out animation"""
+    if not has_lead:
+      self._lead_status_alpha = max(0.0, self._lead_status_alpha - 0.05)
+    else:
+      self._lead_status_alpha = min(1.0, self._lead_status_alpha + 0.1)
+
+  def should_render(self) -> bool:
+    """Check if dev UI should be rendered"""
+    return ui_state.chevron_metrics != ChevronOptions.OFF and self._lead_status_alpha > 0.0
+
+  def _draw_lead(self, lead_data, lead_vehicle, v_ego: float, rect: rl.Rectangle):
+    """Draw lead vehicle status information (distance, speed, TTC)"""
+    if not self.should_render():
+      return
+
+    d_rel = lead_data.dRel
+    v_rel = lead_data.vRel
+
+    if not lead_vehicle.chevron or len(lead_vehicle.chevron) < 2:
+      return
+
+    chevron_x = lead_vehicle.chevron[1][0]
+    chevron_y = lead_vehicle.chevron[1][1]
+    sz = np.clip((25 * 30) / (d_rel / 3 + 30), 15.0, 30.0) * 2.35
+
+    text_lines = self._build_text_lines(d_rel, v_rel, v_ego)
+    if not text_lines:
+      return
+
+    self._render_text_lines(text_lines, chevron_x, chevron_y, sz, rect)
+
+  @staticmethod
+  def _build_text_lines(d_rel: float, v_rel: float, v_ego: float) -> list[str]:
+    """Build text lines based on chevron info setting"""
+    text_lines = []
+
+    # Distance
+    if ui_state.chevron_metrics == ChevronOptions.DISTANCE_ONLY or ui_state.chevron_metrics == ChevronOptions.ALL:
+      val = max(0.0, d_rel)
+      unit = "m" if ui_state.is_metric else "ft"
+      if not ui_state.is_metric:
+        val *= 3.28084
+      text_lines.append(f"{val:.0f} {unit}")
+
+    # Speed
+    if ui_state.chevron_metrics == ChevronOptions.SPEED_ONLY or ui_state.chevron_metrics == ChevronOptions.ALL:
+      multiplier = CV.MS_TO_KPH if ui_state.is_metric else CV.MS_TO_MPH
+      val = max(0.0, (v_rel + v_ego) * multiplier)
+      unit = "km/h" if ui_state.is_metric else "mph"
+      text_lines.append(f"{val:.0f} {unit}")
+
+    # Time to collision
+    if ui_state.chevron_metrics == ChevronOptions.TTC_ONLY or ui_state.chevron_metrics == ChevronOptions.ALL:
+      val = (d_rel / v_ego) if (d_rel > 0 and v_ego > 0) else 0.0
+      ttc_text = f"{val:.1f} s" if (0 < val < 200) else "---"
+      text_lines.append(ttc_text)
+
+    return text_lines
+
+  def _render_text_lines(self, text_lines: list[str], chevron_x: float, chevron_y: float,
+                         sz: float, rect: rl.Rectangle):
+    """Render text lines with proper centering and positioning"""
+    font_size = 40
+    line_height = 50
+    margin = 20
+
+    text_y = chevron_y + sz + 15
+    total_height = len(text_lines) * line_height
+
+    # Adjust Y position if text would go off screen
+    if text_y + total_height > rect.height - margin:
+      y_max = min(chevron_y, rect.height - margin)
+      text_y = y_max - 15 - total_height
+      text_y = max(margin, text_y)
+
+    alpha = int(255 * self._lead_status_alpha)
+    text_color = rl.Color(255, 255, 255, alpha)
+    shadow_color = rl.Color(0, 0, 0, int(200 * self._lead_status_alpha))
+
+    for i, line in enumerate(text_lines):
+      y = int(text_y + (i * line_height))
+      if y + line_height > rect.height - margin:
+        break
+
+      # Measure actual text width for proper centering
+      text_size = measure_text_cached(self._font, line, font_size, 0)
+      text_width = text_size.x
+
+      # Center the text horizontally on the chevron
+      x = int(chevron_x - text_width / 2)
+      x = int(np.clip(x, margin, rect.width - text_width - margin))
+
+      # Draw shadow
+      rl.draw_text_ex(self._font, line, rl.Vector2(x + 2, y + 2), font_size, 0, shadow_color)
+      # Draw text
+      rl.draw_text_ex(self._font, line, rl.Vector2(x, y), font_size, 0, text_color)
+
+  def draw_lead_status(self, sm, radar_state, rect, lead_vehicles):
+    lead_one = radar_state.leadOne
+    lead_two = radar_state.leadTwo
+
+    has_lead_one = lead_one.status if lead_one else False
+    has_lead_two = lead_two.status if lead_two else False
+
+    self.update_alpha(has_lead_one or has_lead_two)
+
+    if not self.should_render():
+      return
+
+    v_ego = sm['carState'].vEgo
+
+    if has_lead_one and lead_vehicles[0].chevron:
+      self._draw_lead(lead_one, lead_vehicles[0], v_ego, rect)
+
+    if has_lead_two and lead_vehicles[1].chevron:
+      d_rel_diff = abs(lead_one.dRel - lead_two.dRel) if has_lead_one else float('inf')
+      if d_rel_diff > 3.0:
+        self._draw_lead(lead_two, lead_vehicles[1], v_ego, rect)

selfdrive/ui/sunnypilot/onroad/model_renderer.py

@@ -0,0 +1,14 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+from openpilot.selfdrive.ui.sunnypilot.onroad.chevron_metrics import ChevronMetrics
+from openpilot.selfdrive.ui.sunnypilot.onroad.rainbow_path import RainbowPath
+
+
+class ModelRendererSP:
+  def __init__(self):
+    self.rainbow_path = RainbowPath()
+    self.chevron_metrics = ChevronMetrics()

selfdrive/ui/sunnypilot/onroad/rainbow_path.py

@@ -0,0 +1,78 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+import time
+import colorsys
+import pyray as rl
+from openpilot.system.ui.lib.shader_polygon import draw_polygon, Gradient
+
+
+class RainbowPath:
+  DEFAULT_NUM_SEGMENTS = 8
+  DEFAULT_SPEED = 50.0  # degrees per second
+  DEFAULT_SATURATION = 0.9
+  DEFAULT_LIGHTNESS = 0.6
+  BASE_ALPHA = 0.8
+  ALPHA_FADE = 0.3  # Alpha reduction from bottom to top
+
+  def __init__(self, num_segments: int = None, speed: float = None, saturation: float = None, lightness: float = None):
+    self.num_segments = num_segments if num_segments is not None else self.DEFAULT_NUM_SEGMENTS
+    self.speed = speed if speed is not None else self.DEFAULT_SPEED
+    self.saturation = saturation if saturation is not None else self.DEFAULT_SATURATION
+    self.lightness = lightness if lightness is not None else self.DEFAULT_LIGHTNESS
+
+  def set_speed(self, speed: float):
+    self.speed = speed
+
+  def set_num_segments(self, num_segments: int):
+    self.num_segments = num_segments
+
+  def set_saturation(self, saturation: float):
+    self.saturation = max(0.0, min(1.0, saturation))
+
+  def set_lightness(self, lightness: float):
+    self.lightness = max(0.0, min(1.0, lightness))
+
+  def get_gradient(self) -> Gradient:
+    time_offset = time.monotonic()
+    hue_offset = (time_offset * self.speed) % 360.0
+
+    segment_colors = []
+    gradient_stops = []
+
+    for i in range(self.num_segments):
+      position = i / (self.num_segments - 1)
+      hue = (hue_offset + position * 360.0) % 360.0
+      alpha = self.BASE_ALPHA * (1.0 - position * self.ALPHA_FADE)
+      color = self._hsla_to_color(
+        hue / 360.0,
+        self.saturation,
+        self.lightness,
+        alpha
+      )
+      gradient_stops.append(position)
+      segment_colors.append(color)
+
+    return Gradient(
+      start=(0.0, 1.0),  # Bottom of path
+      end=(0.0, 0.0),  # Top of path
+      colors=segment_colors,
+      stops=gradient_stops,
+    )
+
+  @staticmethod
+  def _hsla_to_color(h: float, s: float, l: float, a: float) -> rl.Color:
+    rgb = colorsys.hls_to_rgb(h, l, s)
+    return rl.Color(
+      int(rgb[0] * 255),
+      int(rgb[1] * 255),
+      int(rgb[2] * 255),
+      int(a * 255)
+    )
+
+  def draw_rainbow_path(self, rect, path):
+    gradient = self.get_gradient()
+    draw_polygon(rect, path.projected_points, gradient=gradient)

selfdrive/ui/sunnypilot/ui_state.py

@@ -4,10 +4,13 @@ Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
 This file is part of sunnypilot and is licensed under the MIT License.
 See the LICENSE.md file in the root directory for more details.
 """
-from cereal import messaging, custom
+from cereal import messaging, log, custom
 from openpilot.common.params import Params
 from openpilot.sunnypilot.sunnylink.sunnylink_state import SunnylinkState
 
+OpenpilotState = log.SelfdriveState.OpenpilotState
+MADSState = custom.ModularAssistiveDrivingSystem.ModularAssistiveDrivingSystemState
+
 
 class UIStateSP:
   def __init__(self):
@@ -22,9 +25,48 @@ class UIStateSP:
   def update(self) -> None:
     self.sunnylink_state.start()
 
+  @staticmethod
+  def update_status(ss, ss_sp, onroad_evt) -> str:
+    state = ss.state
+    mads = ss_sp.mads
+    mads_state = mads.state
+
+    if state == OpenpilotState.preEnabled:
+      return "override"
+
+    if state == OpenpilotState.overriding:
+      if not mads.available:
+        return "override"
+
+      if any(e.overrideLongitudinal for e in onroad_evt):
+        return "override"
+
+    if mads_state in (MADSState.paused, MADSState.overriding):
+      return "override"
+
+    # MADS specific statuses
+    if not mads.available:
+      return "engaged" if ss.enabled else "disengaged"
+
+    if not mads.enabled and not ss.enabled:
+      return "disengaged"
+
+    if mads.enabled and ss.enabled:
+      return "engaged"
+
+    if mads.enabled:
+      return "lat_only"
+
+    if ss.enabled:
+      return "long_only"
+
+    return "disengaged"
+
   def update_params(self) -> None:
     CP_SP_bytes = self.params.get("CarParamsSPPersistent")
     if CP_SP_bytes is not None:
       self.CP_SP = messaging.log_from_bytes(CP_SP_bytes, custom.CarParamsSP)
     self.sunnylink_enabled = self.params.get_bool("SunnylinkEnabled")
     self.developer_ui = self.params.get("DevUIInfo")
+    self.rainbow_path = self.params.get_bool("RainbowMode")
+    self.chevron_metrics = self.params.get("ChevronInfo")

selfdrive/ui/ui_state.py

@@ -21,6 +21,8 @@ class UIStatus(Enum):
   DISENGAGED = "disengaged"
   ENGAGED = "engaged"
   OVERRIDE = "override"
+  LAT_ONLY = "lat_only"
+  LONG_ONLY = "long_only"
 
 
 class UIState(UIStateSP):
@@ -169,6 +171,8 @@ class UIState(UIStateSP):
       # check for alert
       self.has_alert = True if ss.alertSize != 0 else False
 
+      self.status = UIStatus(UIStateSP.update_status(ss, self.sm["selfdriveStateSP"], self.sm["onroadEvents"]))
+
     # Check for engagement state changes
     if self.engaged != self._engaged_prev:
       for callback in self._engaged_transition_callbacks:

sunnypilot/sunnylink/athena/sunnylinkd.py

@@ -42,6 +42,12 @@ METADATA_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__f
 
 params = Params()
 
+# Parameters that should never be remotely modified for security reasons
+BLOCKED_PARAMS = {
+  "GithubUsername",  # Could grant SSH access
+  "GithubSshKeys",   # Direct SSH key injection
+}
+
 
 def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None:
   cloudlog.info("sunnylinkd.handle_long_poll started")
@@ -248,6 +254,11 @@ def getParams(params_keys: list[str], compression: bool = False) -> str | dict[s
 @dispatcher.add_method
 def saveParams(params_to_update: dict[str, str], compression: bool = False) -> None:
   for key, value in params_to_update.items():
+    # disallow modifications to blocked parameters
+    if key in BLOCKED_PARAMS:
+      cloudlog.warning(f"sunnylinkd.saveParams.blocked: Attempted to modify blocked parameter '{key}'")
+      continue
+
     try:
       save_param_from_base64_encoded_string(key, value, compression)
     except Exception as e:

sunnypilot/sunnylink/athena/tests/test_sunnylinkd.py

@@ -0,0 +1,59 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+from openpilot.sunnypilot.sunnylink.athena import sunnylinkd
+
+
+class TestSunnylinkdMethods:
+  def setup_method(self):
+    self.saved_params = []
+
+    self.original_save = sunnylinkd.save_param_from_base64_encoded_string
+
+    def mock_save_param(key, value, compression=False):
+      self.saved_params.append((key, value, compression))
+
+    sunnylinkd.save_param_from_base64_encoded_string = mock_save_param
+
+  def teardown_method(self):
+    sunnylinkd.save_param_from_base64_encoded_string = self.original_save
+
+  def test_saveParams_blocked(self):
+    blocked_params = {
+      "GithubUsername": "attacker",
+      "GithubSshKeys": "ssh-rsa attacker_key",
+    }
+
+    sunnylinkd.saveParams(blocked_params)
+
+    assert len(self.saved_params) == 0
+
+  def test_saveParams_allowed(self):
+    allowed_params = {
+      "SpeedLimitOffset": "5",
+      "MyCustomParam": "123"
+    }
+
+    sunnylinkd.saveParams(allowed_params)
+
+    # verify content
+    assert len(self.saved_params) == 2
+    keys_saved = [p[0] for p in self.saved_params]
+    assert "SpeedLimitOffset" in keys_saved
+    assert "MyCustomParam" in keys_saved
+
+  def test_saveParams_mixed(self):
+    mixed_params = {
+      "GithubUsername": "attacker",
+      "SpeedLimitOffset": "10"
+    }
+
+    sunnylinkd.saveParams(mixed_params)
+
+    # should save allowed one
+    assert len(self.saved_params) == 1
+    assert self.saved_params[0][0] == "SpeedLimitOffset"
+    assert self.saved_params[0][1] == "10"

sunnypilot/sunnylink/backups/manager.py

@@ -19,7 +19,7 @@ from openpilot.system.version import get_version
 
 from cereal import messaging, custom
 from openpilot.sunnypilot.sunnylink.api import SunnylinkApi
-from openpilot.sunnypilot.sunnylink.backups.utils import decrypt_compressed_data, encrypt_compress_data, SnakeCaseEncoder
+from openpilot.sunnypilot.sunnylink.backups.utils import decrypt_compressed_data, encrypt_compressed_data, SnakeCaseEncoder
 from openpilot.sunnypilot.sunnylink.utils import get_param_as_byte, save_param_from_base64_encoded_string
 
 
@@ -95,7 +95,7 @@ class BackupManagerSP:
 
       # Serialize and encrypt config data
       config_json = json.dumps(config_data)
-      encrypted_config = encrypt_compress_data(config_json, use_aes_256=True)
+      encrypted_config = encrypt_compressed_data(config_json, use_aes_256=True)
       self._update_progress(50.0, OperationType.BACKUP)
 
       backup_info = custom.BackupManagerSP.BackupInfo()

sunnypilot/sunnylink/backups/utils.py

@@ -4,9 +4,9 @@ Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
 This file is part of sunnypilot and is licensed under the MIT License.
 See the LICENSE.md file in the root directory for more details.
 """
-
 import base64
 import hashlib
+import os
 import zlib
 import re
 import json
@@ -14,8 +14,9 @@ from pathlib import Path
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric import rsa, ec
 
+from openpilot.common.api.base import KEYS
 from openpilot.sunnypilot.sunnylink.backups.AESCipher import AESCipher
 from openpilot.system.hardware.hw import Paths
 
@@ -27,37 +28,43 @@ class KeyDerivation:
       return f.read()
 
   @staticmethod
-  def derive_aes_key_iv_from_rsa(key_path: str, use_aes_256: bool) -> tuple[bytes, bytes]:
-    rsa_key_pem: bytes = KeyDerivation._load_key(key_path)
-    key_plain = rsa_key_pem.decode(errors="ignore")
+  def derive_aes_key_iv(key_path: str, use_aes_256: bool) -> tuple[bytes, bytes]:
+    key_pem: bytes = KeyDerivation._load_key(key_path)
+    key_plain = key_pem.decode(errors="ignore")
 
     if "private" in key_plain.lower():
-      private_key = serialization.load_pem_private_key(rsa_key_pem, password=None, backend=default_backend())
-      if not isinstance(private_key, rsa.RSAPrivateKey):
-        raise ValueError("Invalid RSA key format: Unable to determine if key is public or private.")
-
-      der_data = private_key.private_bytes(
-        encoding=serialization.Encoding.DER,
-        format=serialization.PrivateFormat.TraditionalOpenSSL,
-        encryption_algorithm=serialization.NoEncryption()
-      )
+      private_key = serialization.load_pem_private_key(key_pem, password=None, backend=default_backend())
+      if isinstance(private_key, (rsa.RSAPrivateKey, ec.EllipticCurvePrivateKey)):
+        public_key = private_key.public_key()
+      else:
+        raise ValueError("Invalid key format: Unable to determine if key is public or private.")
     elif "public" in key_plain.lower():
-      public_key = serialization.load_pem_public_key(rsa_key_pem, backend=default_backend())
-      if not isinstance(public_key, rsa.RSAPublicKey):
-        raise ValueError("Invalid RSA key format: Unable to determine if key is public or private.")
-
-      der_data = public_key.public_bytes(encoding=serialization.Encoding.DER, format=serialization.PublicFormat.PKCS1)
+      public_key = serialization.load_pem_public_key(key_pem, backend=default_backend())  # type: ignore[assignment]
+      if not isinstance(public_key, (rsa.RSAPublicKey, ec.EllipticCurvePublicKey)):
+        raise ValueError("Invalid key format: Unable to determine if key is public or private.")
     else:
-      raise ValueError("Unknown key format: Unable to determine if key is public or private.")
+      raise ValueError("Invalid key format: Unable to determine if key is public or private.")
 
-    sha256_hash = hashlib.sha256(der_data).digest()
-    aes_key = sha256_hash[:32] if use_aes_256 else sha256_hash[:16]
-    aes_iv = sha256_hash[16:32]
+    if isinstance(public_key, rsa.RSAPublicKey):
+      der_data = public_key.public_bytes(encoding=serialization.Encoding.DER, format=serialization.PublicFormat.PKCS1)
+    elif isinstance(public_key, ec.EllipticCurvePublicKey):
+      der_data = public_key.public_bytes(encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo)
+    else:
+      raise ValueError("Unsupported key type.")
 
-    return aes_key, aes_iv
+    if use_aes_256:
+      # AES-256-CBC
+      key = hashlib.sha256(der_data).digest()
+      iv = hashlib.md5(der_data).digest()
+    else:
+      # AES-128-CBC
+      key = hashlib.md5(der_data).digest()
+      iv = hashlib.md5(der_data).digest()  # Insecure IV reuse, kept for compatibility
+
+    return key, iv
 
 
-def qUncompress(data):
+def uncompress_dat(data):
   """
   Decompress data using zlib.
 
@@ -71,7 +78,7 @@ def qUncompress(data):
   return zlib.decompress(data_stripped_4)
 
 
-def qCompress(data):
+def compress_dat(data):
   """
   Compress data using zlib.
 
@@ -85,6 +92,19 @@ def qCompress(data):
   return b"ZLIB" + compressed_data
 
 
+def get_key_path(use_aes_256=False) -> str:
+  key_path = ""
+  for key in KEYS:
+    if os.path.isfile(Paths.persist_root() + f'/comma/{key}') and os.path.isfile(Paths.persist_root() + f'/comma/{key}.pub'):
+      key_path = str(Path(Paths.persist_root() + f'/comma/{key}') if use_aes_256 else Path(Paths.persist_root() + f'/comma/{key}.pub'))
+      break
+
+  if not key_path:
+    raise FileNotFoundError("No valid key pair found in persist storage.")
+
+  return key_path
+
+
 def decrypt_compressed_data(encrypted_base64, use_aes_256=False):
   """
   Decrypt and decompress data from base64 string.
@@ -96,18 +116,17 @@ def decrypt_compressed_data(encrypted_base64, use_aes_256=False):
   Returns:
       str: Decrypted and decompressed string
   """
-  key_path = Path(f"{Paths.persist_root()}/comma/id_rsa") if use_aes_256 else Path(f"{Paths.persist_root()}/comma/id_rsa.pub")
   try:
     # Decode base64
     encrypted_data = base64.b64decode(encrypted_base64)
 
     # Decrypt
-    key, iv = KeyDerivation.derive_aes_key_iv_from_rsa(str(key_path), use_aes_256)
+    key, iv = KeyDerivation.derive_aes_key_iv(get_key_path(use_aes_256), use_aes_256)
     cipher = AESCipher(key, iv)
     decrypted_data = cipher.decrypt(encrypted_data)
 
     # Decompress
-    decompressed_data = qUncompress(decrypted_data)
+    decompressed_data = uncompress_dat(decrypted_data)
 
     # Decode UTF-8
     result = decompressed_data.decode('utf-8')
@@ -117,7 +136,7 @@ def decrypt_compressed_data(encrypted_base64, use_aes_256=False):
     return ""
 
 
-def encrypt_compress_data(text, use_aes_256=True):
+def encrypt_compressed_data(text, use_aes_256=True):
   """
   Compress and encrypt string data to base64.
 
@@ -128,16 +147,15 @@ def encrypt_compress_data(text, use_aes_256=True):
   Returns:
       str: Base64 encoded encrypted data
   """
-  key_path = Path(f"{Paths.persist_root()}/comma/id_rsa") if use_aes_256 else Path(f"{Paths.persist_root()}/comma/id_rsa.pub")
   try:
     # Encode to UTF-8
     text_bytes = text.encode('utf-8')
 
     # Compress
-    compressed_data = qCompress(text_bytes)
+    compressed_data = compress_dat(text_bytes)
 
     # Encrypt
-    key, iv = KeyDerivation.derive_aes_key_iv_from_rsa(str(key_path), use_aes_256)
+    key, iv = KeyDerivation.derive_aes_key_iv(get_key_path(use_aes_256), use_aes_256)
     cipher = AESCipher(key, iv)
     encrypted_data = cipher.encrypt(compressed_data)