athena: validate all rpc params (#39190)

openpilot/system/athena/athenad.py

@@ -48,7 +48,8 @@ from openpilot.tools.lib.helpers import RE
 from openpilot.common.swaglog import cloudlog
 from openpilot.common.version import get_build_metadata
 from openpilot.common.hardware.hw import Paths
-from openpilot.system.athena.rpc import dispatcher, dumps_call, handle, is_call, is_response, loads
+from openpilot.system.athena.rpc import (boolean, dict_of, dispatcher, dumps_call, either, handle, integer, is_call, is_response,
+                                         list_of, loads, number, one_of, regex, struct)
 
 
 from openpilot.starpilot.connect.provider import active_provider, owns_recording, cloudlog_root, mark_recording
@@ -85,6 +86,15 @@ UploadItemDict = dict[str, str | bool | int | float | dict[str, str]]
 
 UploadFilesToUrlResponse = dict[str, int | list[UploadItemDict] | list[str]]
 
+# param validators for RPC inputs
+DATA_PATH = r'(?!.*\.\.)[\w\-.][\w\-./]*'  # relative to log root
+UPLOAD_URL = regex(r'https?://[\x21-\x7e]+')
+UPLOAD_HEADERS = dict_of(regex(r'[\w\-]+'), regex(r'[\x20-\x7e]*'))
+UPLOAD_ID = regex(r'[0-9a-f]{40}')
+CLIP_FILENAME = regex(r'[\w\-][\w\-.]*')
+UPLOAD_FILE = struct({"fn": regex(DATA_PATH), "url": UPLOAD_URL, "headers": UPLOAD_HEADERS},
+                     {"allow_cellular": boolean, "priority": integer(0, DEFAULT_UPLOAD_PRIORITY)})
+
 
 class UploadTOSAdapter(HTTPAdapter):
   def init_poolmanager(self, connections, maxsize, block=DEFAULT_POOLBLOCK, **pool_kwargs):
@@ -140,7 +150,6 @@ class UploadItem:
     return self.priority == other.priority
 
 
-dispatcher["echo"] = lambda s: s
 recv_queue: Queue[str] = queue.Queue()
 send_queue: Queue[tuple[int, int, str]] = queue.PriorityQueue()
 upload_queue: Queue[UploadItem] = queue.PriorityQueue()
@@ -220,7 +229,8 @@ def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None:
 
 
 def jsonrpc_handler(end_event: threading.Event) -> None:
-  dispatcher["startLocalProxy"] = partial(startLocalProxy, end_event)
+  dispatcher.add_method(partial(startLocalProxy, end_event), name="startLocalProxy",
+                        remote_ws_uri=regex(r'wss://ssh\.comma\.ai(:(443|8443))?/[0-9a-f]{16}'), local_port=one_of(8022, *LOCAL_PORT_WHITELIST))
   while not end_event.is_set():
     try:
       data = recv_queue.get(timeout=1)
@@ -358,11 +368,8 @@ def _do_upload(upload_item: UploadItem, callback: Callable | None = None) -> req
 
 
 # security: user should be able to request any message from their car
-@dispatcher.add_method
+@dispatcher.add_method(service=one_of(*SERVICE_LIST), timeout=integer(0, 10_000))
 def getMessage(service: str, timeout: int = 1000) -> dict:
-  if service is None or service not in SERVICE_LIST:
-    raise Exception("invalid service")
-
   socket = messaging.sub_sock(service, timeout=timeout)
   try:
     ret = messaging.recv_one(socket)
@@ -409,7 +416,7 @@ def scan_dir(path: str, prefix: str) -> list[str]:
           files.append(rel_path)
   return files
 
-@dispatcher.add_method
+@dispatcher.add_method(prefix=either(one_of(''), regex(DATA_PATH)))
 def listDataDirectory(prefix='') -> list[str]:
   return scan_dir(Paths.log_root(), prefix)
 
@@ -633,13 +640,16 @@ class VideoClips:
 
 
 video_clips = VideoClips()
-dispatcher.add_method(video_clips.createClip)
-dispatcher.add_method(video_clips.getClipState)
-dispatcher.add_method(video_clips.deleteClip)
-dispatcher.add_method(video_clips.getClipChunk)
+dispatcher.add_method(video_clips.createClip, route=regex(RE.ROUTE_NAME),
+                      source_start_time=number(0, 24*3600), source_end_time=number(0, 24*3600),
+                      clip=struct({"camera": regex(r'[a-z]*camera\.hevc'), "filename": CLIP_FILENAME, "bitrate": number(1, 20),
+                                   "speedup": integer(1, 10)}))
+dispatcher.add_method(video_clips.getClipState, route=regex(RE.ROUTE_NAME))
+dispatcher.add_method(video_clips.deleteClip, filename=CLIP_FILENAME)
+dispatcher.add_method(video_clips.getClipChunk, filename=CLIP_FILENAME, offset=integer(0, 2**32))
 
 
-@dispatcher.add_method
+@dispatcher.add_method(fn=regex(DATA_PATH), url=UPLOAD_URL, headers=UPLOAD_HEADERS)
 def uploadFileToUrl(fn: str, url: str, headers: dict[str, str]) -> UploadFilesToUrlResponse:
   # this is because mypy doesn't understand that the decorator doesn't change the return type
   response: UploadFilesToUrlResponse = uploadFilesToUrls([{
@@ -650,17 +660,13 @@ def uploadFileToUrl(fn: str, url: str, headers: dict[str, str]) -> UploadFilesTo
   return response
 
 
-@dispatcher.add_method
+@dispatcher.add_method(files_data=list_of(UPLOAD_FILE))
 def uploadFilesToUrls(files_data: list[UploadFileDict]) -> UploadFilesToUrlResponse:
   files = map(UploadFile.from_dict, files_data)
 
   items: list[UploadItemDict] = []
   failed: list[str] = []
   for file in files:
-    if len(file.fn) == 0 or file.fn[0] == '/' or '..' in file.fn or len(file.url) == 0:
-      failed.append(file.fn)
-      continue
-
     path = os.path.join(Paths.log_root(), file.fn)
     if not owns_recording(path, Paths.log_root()) or (not os.path.exists(path) and not os.path.exists(strip_zst_extension(path))):
       failed.append(file.fn)
@@ -701,7 +707,7 @@ def listUploadQueue() -> list[UploadItemDict]:
   return [asdict(i) for i in items if (i is not None) and (i.id not in cancelled_uploads)]
 
 
-@dispatcher.add_method
+@dispatcher.add_method(upload_id=either(UPLOAD_ID, list_of(UPLOAD_ID)))
 def cancelUpload(upload_id: str | list[str]) -> dict[str, int | str]:
   if not isinstance(upload_id, list):
     upload_id = [upload_id]
@@ -714,10 +720,8 @@ def cancelUpload(upload_id: str | list[str]) -> dict[str, int | str]:
   cancelled_uploads.update(cancelled_ids)
   return {"success": 1}
 
-@dispatcher.add_method
+@dispatcher.add_method(latitude=number(-90, 90), longitude=number(-180, 180))
 def setNavDestination(latitude: float, longitude: float) -> dict[str, int | str]:
-  if not (-90 <= latitude <= 90 and -180 <= longitude <= 180):
-    return {"success": 0, "error": "Invalid destination coordinates"}
   Params().put("NavDestination", {"latitude": latitude, "longitude": longitude}, block=True)
   return {"success": 1}
 
@@ -728,7 +732,7 @@ def clearNavDestination() -> dict[str, int]:
   return {"success": 1}
 
 
-@dispatcher.add_method
+@dispatcher.add_method(route=regex(RE.ROUTE_NAME))
 def setRouteViewed(route: str) -> dict[str, int | str]:
   # maintain a list of the last 10 routes viewed in connect
   params = Params()
@@ -750,15 +754,13 @@ def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local
     if local_port == 8022:
       local_port = 22
 
-    if local_port not in LOCAL_PORT_WHITELIST:
-      raise Exception("Requested local port not whitelisted")
-
     cloudlog.debug("athena.startLocalProxy.starting")
 
     dongle_id = Params().get("DongleId")
     identity_token = Api(dongle_id).get_token()
     ws = create_connection(remote_ws_uri,
                            cookie="jwt=" + identity_token,
+                           redirect_limit=0,
                            enable_multithread=True)
 
     # Set TOS to keep connection responsive while under load.
@@ -825,7 +827,7 @@ def getNetworkMetered() -> bool:
   return HARDWARE.get_network_metered(network_type)
 
 
-@dispatcher.add_method
+@dispatcher.add_method(sdp=regex(r'[\x20-\x7e\r\n]*'), enabled=boolean)
 def startStream(sdp: str, enabled: bool) -> dict:
   from openpilot.system.webrtc.helpers import StreamRequestBody, post_stream_request, wait_for_webrtcd
   params = Params()

openpilot/system/athena/rpc.py

@@ -1,4 +1,6 @@
+import inspect
 import json
+import re
 from collections.abc import Callable, Mapping
 from typing import Any
 
@@ -15,12 +17,62 @@ SERVER_ERROR = -32000
 
 JsonDict = dict[str, Any]
 MethodMap = Mapping[str, Callable[..., Any]]
+Validator = Callable[[Any], bool]
+
+
+# *** param validators ***
+
+def boolean(v: Any) -> bool:
+  return isinstance(v, bool)
+
+def integer(lo: int, hi: int) -> Validator:
+  return lambda v: isinstance(v, int) and not isinstance(v, bool) and lo <= v <= hi
+
+def number(lo: float, hi: float) -> Validator:
+  # also rejects NaN and inf, which json.loads accepts
+  return lambda v: isinstance(v, (int, float)) and not isinstance(v, bool) and lo <= v <= hi
+
+def regex(pattern: str) -> Validator:
+  p = re.compile(pattern, re.ASCII)
+  return lambda v: isinstance(v, str) and p.fullmatch(v) is not None
+
+def one_of(*choices: Any) -> Validator:
+  return lambda v: any(type(v) is type(c) and v == c for c in choices)
+
+def either(*validators: Validator) -> Validator:
+  return lambda v: any(validator(v) for validator in validators)
+
+def list_of(validator: Validator) -> Validator:
+  return lambda v: isinstance(v, list) and all(validator(x) for x in v)
+
+def dict_of(key: Validator, value: Validator) -> Validator:
+  return lambda v: isinstance(v, dict) and all(key(k) and value(x) for k, x in v.items())
+
+def struct(required: dict[str, Validator], optional_fields: dict[str, Validator] | None = None) -> Validator:
+  fields = required | (optional_fields or {})
+  return lambda v: isinstance(v, dict) and required.keys() <= v.keys() <= fields.keys() and all(fields[k](x) for k, x in v.items())
 
 
 class Dispatcher(dict[str, Callable[..., Any]]):
-  def add_method(self, f: Callable[..., Any] | None = None, *, name: str | None = None):
+  """Every param of every method must have a validator, all input is rejected by default."""
+
+  def __setitem__(self, name: str, fn: Callable[..., Any]) -> None:
+    raise TypeError("register methods with add_method")
+
+  def add_method(self, f: Callable[..., Any] | None = None, *, name: str | None = None, **validators: Validator):
     def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
-      self[name or fn.__name__] = fn
+      method = name or fn.__name__
+      sig = inspect.signature(fn)
+      if set(sig.parameters) != set(validators):
+        raise TypeError(f"{method}: params {sorted(sig.parameters)} need exactly one validator each, got {sorted(validators)}")
+
+      def validated(*args, **kwargs):
+        for k, v in sig.bind(*args, **kwargs).arguments.items():
+          if not validators[k](v):
+            raise TypeError(f"invalid param: {k}")
+        return fn(*args, **kwargs)
+
+      super(Dispatcher, self).__setitem__(method, validated)
       return fn
     return decorator(f) if f is not None else decorator
 

openpilot/system/athena/tests/test_athenad.py

@@ -122,16 +122,14 @@ class TestAthenadMethods(OpenpilotTestCase):
 
     # out of range and non-finite coordinates are rejected without touching the destination
     for latitude, longitude in [(91, 0), (0, -181), (float('nan'), 0), (0, float('inf'))]:
-      assert set_destination(latitude, longitude)["success"] == 0, (latitude, longitude)
+      with self.assertRaises(TypeError):
+        set_destination(latitude, longitude)
       assert self.params.get("NavDestination") == destination
 
     assert dispatcher["clearNavDestination"]() == {"success": 1}
     assert dispatcher["clearNavDestination"]() == {"success": 1}
     assert self.params.get("NavDestination") is None
 
-  def test_echo(self):
-    assert dispatcher["echo"]("bob") == "bob"
-
   def test_get_message(self):
     with self.assertRaises(TimeoutError) as _:
       dispatcher["getMessage"]("controlsState", timeout=100)
@@ -350,7 +348,7 @@ class TestAthenadMethods(OpenpilotTestCase):
   @with_upload_handler
   def test_cancel_upload(self):
     item = athenad.UploadItem(path="qlog.zst", url="http://localhost:44444/qlog.zst", headers={},
-                              created_at=int(time.time()*1000), id='id', allow_cellular=True)  # noqa: TID251
+                              created_at=int(time.time()*1000), id='0' * 40, allow_cellular=True)  # noqa: TID251
     athenad.upload_queue.put_nowait(item)
     dispatcher["cancelUpload"](item.id)
 
@@ -497,9 +495,9 @@ class TestAthenadMethods(OpenpilotTestCase):
     thread.start()
     try:
       # with params
-      athenad.recv_queue.put_nowait(json.dumps({"method": "echo", "params": ["hello"], "jsonrpc": "2.0", "id": 0}))
+      athenad.recv_queue.put_nowait(json.dumps({"method": "cancelUpload", "params": ["0" * 40], "jsonrpc": "2.0", "id": 0}))
       _, _, resp = athenad.send_queue.get(timeout=3)
-      assert json.loads(resp) == {'result': 'hello', 'id': 0, 'jsonrpc': '2.0'}
+      assert json.loads(resp) == {'result': {'success': 0, 'error': 'not found'}, 'id': 0, 'jsonrpc': '2.0'}
       # without params
       athenad.recv_queue.put_nowait(json.dumps({"method": "getNetworkType", "jsonrpc": "2.0", "id": 0}))
       _, _, resp = athenad.send_queue.get(timeout=3)