From 751d2369c5fdc3a376f55391a9b289f40df80af1 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:53:18 -0500 Subject: [PATCH] Drive Lifecycle Track drive identity and evidence lifecycle so consumers reset state at the same drive boundaries. --- openpilot/starpilot/drive_state/control.py | 47 +++ openpilot/starpilot/drive_state/evidence.py | 135 ++++++++ openpilot/starpilot/drive_state/owner.py | 157 ++++++++++ openpilot/starpilot/drive_state/resolver.py | 25 ++ .../drive_state/tests/test_evidence.py | 142 +++++++++ .../starpilot/drive_state/tests/test_owner.py | 288 ++++++++++++++++++ 6 files changed, 794 insertions(+) create mode 100644 openpilot/starpilot/drive_state/control.py create mode 100644 openpilot/starpilot/drive_state/evidence.py create mode 100644 openpilot/starpilot/drive_state/owner.py create mode 100644 openpilot/starpilot/drive_state/resolver.py create mode 100644 openpilot/starpilot/drive_state/tests/test_evidence.py create mode 100644 openpilot/starpilot/drive_state/tests/test_owner.py diff --git a/openpilot/starpilot/drive_state/control.py b/openpilot/starpilot/drive_state/control.py new file mode 100644 index 0000000000..ed658fd296 --- /dev/null +++ b/openpilot/starpilot/drive_state/control.py @@ -0,0 +1,47 @@ +"""Local and authenticated presentation of the shared drive-state owner.""" + +import threading +import time + +from .resolver import Mode + + +LABELS = {Mode.AUTO: 'Auto', Mode.OFFROAD: 'Offroad', Mode.ONROAD: 'Onroad'} +CYCLE = (Mode.OFFROAD, Mode.ONROAD, Mode.AUTO) + + +class DriveStateControl: + def __init__(self, owner, physical, *, effective=None, clock=time.monotonic): + self.owner, self.physical, self.effective, self.clock = owner, physical, effective or physical.effective, clock + self.cached = None + self.lock = threading.Lock() + + def snapshot(self): + with self.lock: + now = self.clock() + if self.cached is None or not 0 <= now - self.cached[0] < 0.5: + state = self.owner.snapshot() + status = getattr(self.physical, 'status', None) + admission = status() if callable(status) else {'allowed': self.physical.allowed(), 'reason': None} + self.cached = now, state, bool(admission['allowed']), admission['reason'] + state = self.cached[1] + actual = self.effective() + return {'mode': state.mode.value, 'revision': state.revision, 'available': state.available, + 'effective': 'onroad' if actual is True else 'offroad' if actual is False else None, + 'overrideAllowed': self.cached[2], 'reason': self.cached[3]} + + def change(self, mode, revision, authorized): + # Deliberate local requests may stop driving services while moving; Galaxy rejects Offroad in its action handler. + result = self.owner.request(mode, expected_revision=revision, authorized=authorized, + override_allowed=lambda: True) + with self.lock: + self.cached = None + return result + + @staticmethod + def next_mode(status): + mode = Mode(status['mode']) + return CYCLE[(CYCLE.index(mode) + 1) % len(CYCLE)] + + def cycle(self, status, authorized): + return self.change(self.next_mode(status).value, status['revision'], authorized) diff --git a/openpilot/starpilot/drive_state/evidence.py b/openpilot/starpilot/drive_state/evidence.py new file mode 100644 index 0000000000..153ecc0041 --- /dev/null +++ b/openpilot/starpilot/drive_state/evidence.py @@ -0,0 +1,135 @@ +"""Fresh physical vehicle evidence independent of the requested pipeline mode.""" + +import threading +import time + +from openpilot.starpilot.parked_evidence import DEVICE_TTL_NS, PANDA_TTL_NS, RESUME_SKEW_NS + + +PYTHON_TTL_NS = 300_000_000 + + +class PhysicalSource: + def __init__(self, messages=None, *, owns_messages=False, mono=time.monotonic_ns, + boot=lambda: time.clock_gettime_ns(getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC))): + self.messages, self.owns_messages = messages, owns_messages + self.mono, self.boot = mono, boot + self.floor = getattr(messages, "after_mono_ns", mono()) + self.offset = None + self.lock = threading.Lock() + self.closed = False + + def allowed(self) -> bool: + return self.status()['allowed'] + + def status(self): + def result(allowed, reason=None): + return {'allowed': bool(allowed), 'reason': reason} + with self.lock: + if self.closed: + return result(False, 'Vehicle status source is closed') + try: + if self.messages is None: + from openpilot.cereal import messaging + self.messages = messaging.SubMaster(['deviceState', 'pandaStates', 'carState', 'selfdriveState']) + self.owns_messages = True + snapshot = getattr(self.messages, "snapshot", None) + sm = snapshot() if callable(snapshot) else self.messages + if sm is None: + return result(False, 'Current vehicle status is unavailable') + self.floor = max(self.floor, getattr(sm, "after_mono_ns", self.floor)) + if self.owns_messages: + sm.update(0) + before, boot, now = self.mono(), self.boot(), self.mono() + if not 0 <= now - before <= RESUME_SKEW_NS: + self.floor, self.offset = now, None + return result(False, 'Device clock changed; fresh vehicle status is required') + offset = boot - (before + now) // 2 + if self.offset is not None and abs(offset - self.offset) > RESUME_SKEW_NS: + self.floor = now + self.offset = offset + + def fresh(service, python=False): + stamp = int(sm.logMonoTime[service]) + receipt = int(sm.recv_time[service] * 1e9) + if not all((sm.seen[service], sm.alive[service], sm.valid[service])) or receipt <= self.floor: + return False + if python and stamp <= self.floor: + return False + age = now - stamp if python else boot - stamp + ttl = PYTHON_TTL_NS if python else PANDA_TTL_NS + return stamp > 0 and 0 <= age <= ttl and 0 <= now - receipt <= ttl + + if not fresh('pandaStates'): + return result(False, 'Current Panda status is missing, invalid or older than 300 ms') + pandas = sm['pandaStates'] + if len(pandas) == 0 or any(str(p.pandaType) == 'unknown' for p in pandas): + return result(False, 'No recognized Panda is reporting vehicle status') + ignition = any(p.ignitionLine or p.ignitionCan for p in pandas) + if not ignition: + safe = all(str(p.safetyModel) == 'noOutput' and not p.controlsAllowed for p in pandas) + return result(safe, None if safe else 'Panda must confirm noOutput safety and controls disabled') + if not fresh('carState', True): + return result(False, 'Current speed and gear are unavailable; turn off the vehicle ignition to confirm parked state') + if not fresh('selfdriveState', True): + return result(False, 'Driving services cannot confirm disengagement; turn off the vehicle ignition to confirm parked state') + car, selfdrive = sm['carState'], sm['selfdriveState'] + if not car.canValid or car.canTimeout: + return result(False, 'Vehicle CAN data is invalid or timed out') + if not car.standstill or str(car.gearShifter) != 'park': + return result(False, 'Park the vehicle before forcing Offroad or updating') + if selfdrive.enabled or selfdrive.active: + return result(False, 'Disengage driving controls before forcing Offroad or updating') + return result(True) + except (OSError, RuntimeError, AttributeError, KeyError, TypeError, ValueError, OverflowError): + return result(False, 'Vehicle status could not be read') + + def update_status(self, manager_offroad): + if manager_offroad is not True: + return {'allowed': False, 'reason': 'Switch the device Offroad before updating'} + physical = self.status() + if not physical['allowed']: + return physical + actual = self.effective() + if actual is None: + return {'allowed': False, 'reason': 'Current device offroad status is missing or older than 1.5 seconds'} + if actual: + return {'allowed': False, 'reason': 'Manager is still running driving services; wait for the Offroad transition'} + return {'allowed': True, 'reason': None} + + def update_ready(self, manager_offroad): + return self.update_status(manager_offroad)['allowed'] + + def effective(self): + with self.lock: + if self.closed or self.messages is None or self.offset is None: + return None + try: + before, boot, now = self.mono(), self.boot(), self.mono() + offset = boot - (before + now) // 2 + if not 0 <= now - before <= RESUME_SKEW_NS or abs(offset - self.offset) > RESUME_SKEW_NS: + self.floor, self.offset = now, None + return None + snapshot = getattr(self.messages, "snapshot", None) + sm = snapshot() if callable(snapshot) else self.messages + if sm is None: + return None + self.floor = max(self.floor, getattr(sm, "after_mono_ns", self.floor)) + stamp, receipt = int(sm.logMonoTime['deviceState']), int(sm.recv_time['deviceState'] * 1e9) + if (not all((sm.seen['deviceState'], sm.alive['deviceState'], sm.valid['deviceState'])) or + stamp <= self.floor or receipt <= self.floor or + not 0 <= now - stamp <= DEVICE_TTL_NS or not 0 <= now - receipt <= DEVICE_TTL_NS): + return None + return bool(sm['deviceState'].started) + except (OSError, RuntimeError, AttributeError, KeyError, TypeError, ValueError, OverflowError): + return None + + def close(self): + with self.lock: + self.closed = True + if self.owns_messages and self.messages is not None: + for socket in getattr(self.messages, 'sock', {}).values(): + close = getattr(socket, 'close', None) + if close is not None: + close() + self.messages = None diff --git a/openpilot/starpilot/drive_state/owner.py b/openpilot/starpilot/drive_state/owner.py new file mode 100644 index 0000000000..b7b00b11f4 --- /dev/null +++ b/openpilot/starpilot/drive_state/owner.py @@ -0,0 +1,157 @@ +"""One session-bound request owner shared by local and authenticated controls.""" + +from collections.abc import Callable +from contextlib import contextmanager +from dataclasses import dataclass +import fcntl +import os +from pathlib import Path +import re +import stat +import threading +import uuid + +from openpilot.common.params import UnknownKeyName + +from .resolver import Mode + + +SESSION_KEY = 'DriveStateSession' +REQUEST_KEY = 'DriveStateRequest' +_HEX = re.compile(r'[0-9a-f]{32}\Z') + + +class Rejected(ValueError): + pass + + +@dataclass(frozen=True) +class State: + mode: Mode = Mode.AUTO + revision: str | None = None + available: bool = False + + +def valid_session(value) -> bool: + return (type(value) is dict and set(value) == {'version', 'id', 'pid', 'birth', 'boot'} and + type(value['version']) is int and value['version'] == 1 and + isinstance(value['id'], str) and _HEX.fullmatch(value['id']) is not None and + isinstance(value['boot'], str) and re.fullmatch(r'[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}', value['boot']) is not None and + all(type(value[key]) is int and value[key] > 0 for key in ('pid', 'birth'))) + + +def request_state(value, session) -> State: + if (type(value) is not dict or set(value) != {'version', 'session', 'revision', 'mode'} or + type(value['version']) is not int or value['version'] != 1 or value['session'] != session['id'] or + not isinstance(value['revision'], str) or _HEX.fullmatch(value['revision']) is None): + return State() + try: + return State(Mode(value['mode']), value['revision'], True) + except (TypeError, ValueError): + return State() + + +def process_identity(pid: int): + try: + raw = Path(f"/proc/{pid}/stat").read_bytes() + fields = raw[raw.rfind(b')') + 2:].split() + return int(fields[19]), fields[0] + except (OSError, IndexError, ValueError): + return None, None + + +def manager_alive(session) -> bool: + try: + if Path('/proc/sys/kernel/random/boot_id').read_text().strip() != session['boot']: + return False + except OSError: + return False + birth, state = process_identity(session['pid']) + return birth == session['birth'] and state not in (None, b'Z', b'X') + + +class DriveStateOwner: + def __init__(self, params, root: Path, *, alive: Callable[[dict], bool] = manager_alive): + self.params, self.root, self.alive = params, Path(root), alive + self._lock = threading.RLock() + + @contextmanager + def _exclusive(self): + with self._lock: + self.root.mkdir(mode=0o700, parents=True, exist_ok=True) + info = self.root.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o022: + raise Rejected('Drive state storage is unavailable') + fd = os.open(self.root / '.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077: + raise Rejected('Drive state storage is unavailable') + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + raise Rejected("Drive state changed; refresh and try again") from None + yield + finally: + os.close(fd) + + def _read(self): + try: + session = self.params.get(SESSION_KEY) + if not valid_session(session) or not self.alive(session): + return None, State() + return session, request_state(self.params.get(REQUEST_KEY), session) + except (OSError, KeyError, TypeError, ValueError, UnknownKeyName): + return None, State() + + def snapshot(self) -> State: + return self._read()[1] + + def initialize(self, *, pid: int, birth: int, boot: str) -> State: + session = {'version': 1, 'id': uuid.uuid4().hex, 'pid': pid, 'birth': birth, 'boot': boot} + if not valid_session(session) or not self.alive(session): + raise Rejected('Manager session is unavailable') + with self._exclusive(): + request = self._record(session, Mode.AUTO) + # Mismatched generations resolve Auto during either write or a crash. + self.params.put(REQUEST_KEY, request, block=True) + self.params.put(SESSION_KEY, session, block=True) + return request_state(request, session) + + def initialize_manager(self) -> State: + pid = os.getpid() + boot = Path('/proc/sys/kernel/random/boot_id').read_text().strip() + return self.initialize(pid=pid, birth=process_identity(pid)[0], boot=boot) + + @staticmethod + def _record(session, mode): + return {'version': 1, 'session': session['id'], 'revision': uuid.uuid4().hex, 'mode': mode.value} + + def request(self, mode: str, *, expected_revision: str, authorized: Callable[[], bool], + override_allowed: Callable[[], bool]) -> State: + try: + desired = Mode(mode) + except (TypeError, ValueError): + raise Rejected('Choose Auto, Offroad or Onroad') from None + with self._exclusive(): + session, current = self._read() + if not current.available or current.revision != expected_revision: + raise Rejected('Drive state changed; refresh and try again') + if not authorized(): + raise Rejected('Drive state changes are unavailable') + if desired == Mode.OFFROAD and not override_allowed(): + raise Rejected('Park the vehicle and disengage before forcing Offroad') + if desired == current.mode: + return current + # These callbacks recheck the authenticated caller and physical source, + # rather than relying on the state that produced the button. + if not authorized() or (desired == Mode.OFFROAD and not override_allowed()) or not self.alive(session): + raise Rejected('Drive state changes are unavailable') + observed_session, observed = self._read() + if observed_session != session or observed != current: + raise Rejected('Drive state changed; refresh and try again') + record = self._record(session, desired) + if not authorized() or (desired == Mode.OFFROAD and not override_allowed()): + raise Rejected('Drive state changes are unavailable') + self.params.put(REQUEST_KEY, record, block=True) + return request_state(record, session) diff --git a/openpilot/starpilot/drive_state/resolver.py b/openpilot/starpilot/drive_state/resolver.py new file mode 100644 index 0000000000..45b88b9e26 --- /dev/null +++ b/openpilot/starpilot/drive_state/resolver.py @@ -0,0 +1,25 @@ +"""Pipeline state selection; physical ignition remains separate.""" + +from enum import StrEnum +from collections.abc import Mapping + + +class Mode(StrEnum): + AUTO = 'auto' + OFFROAD = 'offroad' + ONROAD = 'onroad' + + +def effective_onroad(mode: Mode, conditions: Mapping[str, bool]) -> bool: + if mode == Mode.OFFROAD: + return False + if mode == Mode.ONROAD: + if 'ignition' not in conditions: + return False + return all(value for key, value in conditions.items() if key != 'ignition') + return all(conditions.values()) + + +def should_start(mode: Mode, onroad: Mapping[str, bool], startup: Mapping[str, bool], *, already_started: bool) -> bool: + ready = effective_onroad(mode, onroad) + return ready if already_started else ready and all(startup.values()) diff --git a/openpilot/starpilot/drive_state/tests/test_evidence.py b/openpilot/starpilot/drive_state/tests/test_evidence.py new file mode 100644 index 0000000000..c200c7c857 --- /dev/null +++ b/openpilot/starpilot/drive_state/tests/test_evidence.py @@ -0,0 +1,142 @@ +from types import SimpleNamespace as NS +import pytest +from typing import Any +from openpilot.starpilot.drive_state.evidence import PhysicalSource + + +class Messages: + def __init__(self): + self.values: dict[str, Any] = { + 'pandaStates': [NS(pandaType='uno', controlsAllowed=False, ignitionLine=False, ignitionCan=False, safetyModel='noOutput')], + 'carState': NS(canValid=True, canTimeout=False, standstill=True, gearShifter='park'), + 'selfdriveState': NS(enabled=False, active=False), + } + keys = self.values + self.seen = dict.fromkeys(keys, True) + self.alive = dict.fromkeys(keys, True) + self.valid = dict.fromkeys(keys, True) + self.logMonoTime = dict.fromkeys(keys, 2_000_000_000) + self.logMonoTime['pandaStates'] += 10_000_000_000 + self.recv_time = dict.fromkeys(keys, 2.0) + + def __getitem__(self, key): + return self.values[key] + + +@pytest.fixture +def fixture(): + now = [1_000_000_000, 10_000_000_000] + messages = Messages() + source = PhysicalSource(messages, mono=lambda: now[0], boot=lambda: sum(now)) + now[0] = 2_100_000_000 + return NS(now=now, sm=messages, source=source) + + +def test_genuine_ignition_off_needs_all_panda_no_output_not_pipeline_state(fixture): + f = fixture + assert f.source.allowed() + f.sm.values['deviceState'] = NS(started=True) + assert f.source.allowed() + f.sm.values['pandaStates'][0].safetyModel = 'toyota' + assert not f.source.allowed() + + +@pytest.mark.parametrize('field,value', [('controlsAllowed', True), ('pandaType', 'unknown')]) +def test_unknown_or_actuation_authority_never_admits(fixture, field, value): + setattr(fixture.sm.values['pandaStates'][0], field, value) + assert not fixture.source.allowed() + + +def test_live_can_park_disabled_can_admit_with_ignition_on_and_started(fixture): + f = fixture + f.sm.values['pandaStates'][0].ignitionLine = True + f.sm.values['pandaStates'][0].safetyModel = 'toyota' + f.sm.values['deviceState'] = NS(started=True) + assert f.source.allowed() + for message, field, value in [ + ('carState', 'canValid', False), + ('carState', 'canTimeout', True), + ('carState', 'standstill', False), + ('carState', 'gearShifter', 'drive'), + ('selfdriveState', 'enabled', True), + ('selfdriveState', 'active', True), + ]: + before = getattr(f.sm.values[message], field) + setattr(f.sm.values[message], field, value) + assert not f.source.allowed() + setattr(f.sm.values[message], field, before) + + +def test_stale_forced_offroad_cannot_supply_park_authority(fixture): + f = fixture + f.sm.values['pandaStates'][0].ignitionLine = True + f.sm.logMonoTime['carState'] = 1_000_000_000 + f.sm.values['deviceState'] = NS(started=False) + assert not f.source.allowed() + + +@pytest.mark.parametrize('service', ['pandaStates', 'carState', 'selfdriveState']) +def test_missing_invalid_stale_future_sources_fail_closed(fixture, service): + f = fixture + f.sm.values['pandaStates'][0].ignitionLine = True + for table, value in [(f.sm.valid, False), (f.sm.alive, False), (f.sm.seen, False), (f.sm.logMonoTime, 0), (f.sm.recv_time, 100.0)]: + before = table[service] + table[service] = value + assert not f.source.allowed() + table[service] = before + + +def test_resume_and_collector_start_need_post_boundary_evidence(fixture): + f = fixture + f.sm.values['pandaStates'][0].ignitionLine = True + assert f.source.allowed() + f.now[1] += 20_000_000_000 + assert not f.source.allowed() + f.sm.logMonoTime['pandaStates'] = sum(f.now) + f.sm.recv_time['pandaStates'] = f.now[0] / 1e9 + assert not f.source.allowed() + f.now[0] += 10_000_000 + for key in f.sm.values: + if key == 'deviceState': + continue + f.sm.logMonoTime[key] = sum(f.now) if key == 'pandaStates' else f.now[0] + f.sm.recv_time[key] = f.now[0] / 1e9 + assert f.source.allowed() + f.source.close() + assert not f.source.allowed() + + +def test_real_capnp_physical_fields_and_fresh_effective_state(fixture): + from openpilot.cereal import log + + f = fixture + panda_event = log.Event.new_message() + pandas = panda_event.init('pandaStates', 1) + pandas[0].pandaType = 'uno' + pandas[0].safetyModel = 'noOutput' + f.sm.values['pandaStates'] = panda_event.as_reader().pandaStates + assert f.source.allowed() + device = log.Event.new_message().init('deviceState') + device.started = True + f.sm.values['deviceState'] = device.as_reader() + for table, value in ((f.sm.seen, True), (f.sm.alive, True), (f.sm.valid, True), (f.sm.logMonoTime, 2_000_000_000), (f.sm.recv_time, 2.0)): + table['deviceState'] = value + assert f.source.effective() is True + f.sm.logMonoTime['deviceState'] = 1 + assert f.source.effective() is None + f.now[1] += 10_000_000_000 + assert f.source.effective() is None + + +def test_park_disengaged_does_not_confuse_native_permission_with_active_controls(fixture): + f = fixture + f.sm.values['pandaStates'][0].ignitionLine = True + f.sm.values['pandaStates'][0].safetyModel = 'hyundaiCanfd' + f.sm.values['pandaStates'][0].controlsAllowed = True + f.sm.values['deviceState'] = NS(started=True) + assert f.source.allowed() + f.sm.values['selfdriveState'].active = True + assert not f.source.allowed() + f.sm.values['selfdriveState'].active = False + f.sm.values['carState'].gearShifter = 'drive' + assert not f.source.allowed() diff --git a/openpilot/starpilot/drive_state/tests/test_owner.py b/openpilot/starpilot/drive_state/tests/test_owner.py new file mode 100644 index 0000000000..8e37c5d642 --- /dev/null +++ b/openpilot/starpilot/drive_state/tests/test_owner.py @@ -0,0 +1,288 @@ +"""Temporary typed-store fakes exercise the shared request boundary.""" + +import copy +from types import SimpleNamespace as NS + +import pytest + +from openpilot.starpilot.drive_state.owner import DriveStateOwner, Mode, Rejected, REQUEST_KEY, SESSION_KEY + +BOOT = '00000000-0000-0000-0000-000000000001' + + +class Params: + def __init__(self): + self.values = {} + self.writes = [] + + def get(self, key): + return copy.deepcopy(self.values.get(key)) + + def put(self, key, value, block=False): + assert block + self.values[key] = copy.deepcopy(value) + self.writes.append((key, copy.deepcopy(value))) + + +@pytest.fixture +def fixture(tmp_path): + params = Params() + live = [True] + owner = DriveStateOwner(params, tmp_path / 'owner', alive=lambda _: live[0]) + owner.initialize(pid=1, birth=2, boot=BOOT) + return NS(params=params, live=live, owner=owner) + + +def request(f, mode, **kwargs): + return f.owner.request(mode, expected_revision=f.owner.snapshot().revision, + authorized=kwargs.get('authorized', lambda: True), + override_allowed=kwargs.get('override_allowed', lambda: True)) + + +def test_shared_owner_idempotence_atomic_mode_and_manager_reset(fixture): + f = fixture + state = request(f, 'onroad') + writes = len(f.params.writes) + assert request(f, 'onroad') == state and len(f.params.writes) == writes + assert request(f, 'offroad').mode == Mode.OFFROAD + assert set(f.params.values) == {REQUEST_KEY, SESSION_KEY} + old_revision = f.owner.snapshot().revision + f.owner.initialize(pid=1, birth=3, boot=BOOT) + assert f.owner.snapshot().mode == Mode.AUTO + with pytest.raises(Rejected): + f.owner.request('onroad', expected_revision=old_revision, authorized=lambda: True, override_allowed=lambda: True) + + +def test_only_deliberate_override_requires_physical_admission(fixture): + f = fixture + with pytest.raises(Rejected): + request(f, 'offroad', override_allowed=lambda: False) + request(f, 'onroad') + assert request(f, 'auto', override_allowed=lambda: False).mode == Mode.AUTO + assert not any(key in f.params.values for key in ('ForceOffroad', 'ForceOnroad')) + + +@pytest.mark.parametrize('change', ['permission', 'physical', 'manager']) +def test_revocation_before_commit_never_writes_request(fixture, change): + f = fixture + writes = len(f.params.writes) + calls = [0] + + def checked(): + calls[0] += 1 + if calls[0] == 2 and change == 'manager': + f.live[0] = False + return calls[0] < 2 + + with pytest.raises(Rejected): + request(f, 'offroad', authorized=checked if change == 'permission' else lambda: True, + override_allowed=checked if change in ('physical', 'manager') else lambda: True) + assert len(f.params.writes) == writes + + +@pytest.mark.parametrize('damage', ['missing', 'invalid-mode', 'extra-key', 'wrong-generation', 'boolean-version', 'dead']) +def test_bad_or_old_requests_never_grant_override(fixture, damage): + f = fixture + request(f, 'onroad') + if damage == 'missing': + f.params.values.pop(REQUEST_KEY) + elif damage == 'invalid-mode': + f.params.values[REQUEST_KEY]['mode'] = 'both' + elif damage == 'extra-key': + f.params.values[REQUEST_KEY]['ForceOnroad'] = True + elif damage == 'wrong-generation': + f.params.values[REQUEST_KEY]['session'] = '0' * 32 + elif damage == 'boolean-version': + f.params.values[REQUEST_KEY]['version'] = True + else: + f.live[0] = False + assert f.owner.snapshot().mode == Mode.AUTO + assert not f.owner.snapshot().available + + +def test_two_key_initialization_crash_defaults_auto(fixture): + f = fixture + request(f, 'onroad') + old_session = f.params.get(SESSION_KEY) + original_put = f.params.put + + def interrupted(key, value, block=False): + if key == SESSION_KEY: + raise OSError('interrupted manager initialization') + original_put(key, value, block=block) + + f.params.put = interrupted + with pytest.raises(OSError): + f.owner.initialize(pid=1, birth=3, boot=BOOT) + assert f.params.get(SESSION_KEY) == old_session + assert f.owner.snapshot().mode == Mode.AUTO and not f.owner.snapshot().available + + +def test_session_storage_symlink_is_rejected(fixture, tmp_path): + f = fixture + target = tmp_path / 'elsewhere' + target.mkdir() + alias = tmp_path / 'alias' + alias.symlink_to(target) + owner = DriveStateOwner(f.params, alias, alive=lambda _: True) + with pytest.raises(Rejected): + owner.request('onroad', expected_revision=f.owner.snapshot().revision, + authorized=lambda: True, override_allowed=lambda: True) + + +class FileParams: + def __init__(self, root): + self.root = root + + def get(self, key): + import json + try: + return json.loads((self.root / key).read_text()) + except FileNotFoundError: + return None + + def put(self, key, value, block=False): + import json + import os + import uuid + assert block + temporary = self.root / ('.put-' + uuid.uuid4().hex) + temporary.write_text(json.dumps(value)) + os.replace(temporary, self.root / key) + + + +PROCESS_PROGRAM = """ +import json,sys,time,types +from pathlib import Path +sys.path.insert(0,sys.argv[2]) +sys.path.insert(0,sys.argv[1]) +import openpilot.starpilot +pkg=types.ModuleType('openpilot.starpilot.drive_state') +pkg.__path__=[str(Path(sys.argv[1])/'openpilot/starpilot/drive_state')] +sys.modules[pkg.__name__]=pkg +from openpilot.starpilot.drive_state.owner import DriveStateOwner, Rejected +# FILE_PARAMS +params_root,lock_root,revision,mode,ready,release,result,started,go=sys.argv[3:] +owner=DriveStateOwner(FileParams(Path(params_root)),Path(lock_root),alive=lambda _: True) +if started != '-': + Path(started).touch() + startup_deadline=time.monotonic()+15 + while not Path(go).exists(): + if time.monotonic()>startup_deadline:raise RuntimeError('request barrier timeout') + time.sleep(.01) +def authorized(): + if ready != '-': + Path(ready).touch() + deadline=time.monotonic()+30 + while not Path(release).exists(): + if time.monotonic()>deadline:raise RuntimeError('fixture timeout') + time.sleep(.01) + return True +try: + state=owner.request(mode,expected_revision=revision,authorized=authorized,override_allowed=lambda: True) + output=['accepted',state.mode.value] +except Rejected: + output=['rejected',mode] +Path(result).write_text(json.dumps(output)) +""" + + +def test_crossprocess_revision_commit_and_nonblocking_hardware_snapshot(tmp_path): + import json + import subprocess + import sys + import inspect + import time + from pathlib import Path + params_root = tmp_path / 'params' + params_root.mkdir() + lock_root = tmp_path / 'owner' + owner = DriveStateOwner(FileParams(params_root), lock_root, alive=lambda _: True) + state = owner.initialize(pid=1, birth=2, boot=BOOT) + ready, release = tmp_path / 'ready', tmp_path / 'release' + source = Path(__file__).parents[4] + from openpilot.common import params as native_params + native_source = Path(native_params.__file__).parents[2] + result1, result2 = tmp_path / 'result1', tmp_path / 'result2' + program = PROCESS_PROGRAM.replace('# FILE_PARAMS', inspect.getsource(FileParams)) + common = [sys.executable, '-c', program, str(source), str(native_source), str(params_root), str(lock_root), state.revision] + assert state.revision is not None + common = [str(value) for value in common] + first = subprocess.Popen(common + ['onroad', str(ready), str(release), str(result1), '-', '-']) + second = None + second_ready, second_go = tmp_path / 'second-ready', tmp_path / 'second-go' + try: + deadline = time.monotonic() + 15 + while not ready.exists() and time.monotonic() < deadline: + time.sleep(.01) + assert ready.exists() + # Hardware reads do not join a UI/auth request holding the writer lock. + assert owner.snapshot() == state + assert not release.exists() + second = subprocess.Popen(common + ['offroad', '-', str(release), str(result2), str(second_ready), str(second_go)]) + # Interpreter/native imports have their own startup budget, separate from + # the writer latency proof. The holder is never released by a timer. + deadline = time.monotonic() + 15 + while not second_ready.exists() and time.monotonic() < deadline: + time.sleep(.01) + assert second_ready.exists() and not release.exists() + second_go.touch() + assert second.wait(timeout=2) == 0 + assert not release.exists() # A competing UI request fails without waiting on the holder. + release.touch() + assert first.wait(timeout=5) == 0 + assert sorted((json.loads(result1.read_text()), json.loads(result2.read_text()))) == [ + ['accepted', 'onroad'], ['rejected', 'offroad']] + assert owner.snapshot().mode == Mode.ONROAD + finally: + release.touch() + for process in (first, second): + if process is not None and process.poll() is None: + process.kill() + process.wait(timeout=3) + + +def test_process_identity_and_boot_liveness_use_single_stat_read(monkeypatch): + from pathlib import Path + from openpilot.starpilot.drive_state.owner import process_identity, manager_alive + reads = [] + # Linux field 22 follows comm (which may contain spaces and closing parentheses). + fields = [b'S'] + [b'0'] * 18 + [b'12345'] + raw = b'42 (manager ) worker) ' + b' '.join(fields) + def read_bytes(path): + reads.append(str(path)) + return raw + monkeypatch.setattr(Path, 'read_bytes', read_bytes) + monkeypatch.setattr(Path, 'read_text', lambda path: BOOT) + session = {'pid': 42, 'birth': 12345, 'boot': BOOT} + assert manager_alive(session) + assert reads == ['/proc/42/stat'] + assert not manager_alive(dict(session, birth=12346)) + assert not manager_alive(dict(session, boot='different-boot')) + raw = raw.replace(b') S ', b') Z ') + assert not manager_alive(session) + raw = b'malformed' + assert process_identity(42) == (None, None) + + +def test_empty_real_native_registry_resolves_auto_and_unsupported_key_is_explicit(tmp_path): + from openpilot.common.params import Params as NativeParams, UnknownKeyName + from openpilot.starpilot.drive_state.owner import State + params = NativeParams(str(tmp_path / 'params')) + owner = DriveStateOwner(params, tmp_path / 'owner') + assert owner.snapshot() == State() + class Unsupported: + def get(self, key): + raise UnknownKeyName(key) + assert DriveStateOwner(Unsupported(), tmp_path / 'unsupported').snapshot() == State() + + +def test_force_onroad_requires_no_park_proof_but_offroad_does(fixture): + f = fixture + assert request(f, 'onroad', override_allowed=lambda: False).mode == Mode.ONROAD + with pytest.raises(Rejected): + request(f, 'offroad', override_allowed=lambda: False) + with pytest.raises(Rejected): + request(f, 'auto', authorized=lambda: False, override_allowed=lambda: False) + assert f.owner.snapshot().mode == Mode.ONROAD