diff --git a/openpilot/system/athena/athenad.py b/openpilot/system/athena/athenad.py index 36c55d54d8..c4fdd7461e 100755 --- a/openpilot/system/athena/athenad.py +++ b/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() diff --git a/openpilot/system/athena/rpc.py b/openpilot/system/athena/rpc.py index aaf7f41fc3..922f8c733a 100644 --- a/openpilot/system/athena/rpc.py +++ b/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 diff --git a/openpilot/system/athena/tests/test_athenad.py b/openpilot/system/athena/tests/test_athenad.py index 486da2d304..271b6405c1 100644 --- a/openpilot/system/athena/tests/test_athenad.py +++ b/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)