diff --git a/roadscore/prototype/passive_observer_policy.py b/roadscore/prototype/passive_observer_policy.py new file mode 100644 index 0000000000..47f118e62b --- /dev/null +++ b/roadscore/prototype/passive_observer_policy.py @@ -0,0 +1,74 @@ +"""Read-only passive safety evidence, not vehicle authorization or live readiness. + +The collector must still enforce calibration, camera/model validity, local model +placement, CPU/thermal/link bounds, parked coexistence and explicit driver ready. +No Params, control messages, safety modes, or engagement state are modified here. +""" +import hashlib +import json +import math + +MODE = 'passive-observer-v1' + + +def _mapping(value): + return value if isinstance(value,dict) else {} + + +def _fresh(snapshot, topic, maximum): + age=_mapping(snapshot.get('ages')).get(topic) + return (_mapping(snapshot.get('valid')).get(topic) is True and type(age) in (int,float) + and math.isfinite(age) and 0<=age<=maximum) + + +def evaluate_passive_observer(snapshot, requested_mode): + """Accept only actual nonactuating passive evidence; never infer it from a toggle. + + snapshot adds car_identity.passive/notCar/dashcamOnly to the collector's + fingerprint/firmware/safety identity, plus raw carControl/selfdriveState dicts + as `control`/`selfdrive_state`. Freshness uses the collector's source ages. + """ + snapshot=_mapping(snapshot) + reasons=[] + if requested_mode!=MODE:reasons.append('Explicit passive observer mode is required') + cp=_mapping(snapshot.get('car_identity')) + if cp.get('passive') is not True:reasons.append('Actual CarParams must be passive') + if cp.get('notCar') is not False:reasons.append('Actual recognized-car mode is required') + if type(cp.get('dashcamOnly')) is not bool:reasons.append('Actual dashcamOnly metadata is missing') + configs=cp.get('safety') + if not isinstance(configs,list) or not configs or any(not isinstance(s,dict) or s.get('model')!='noOutput' or type(s.get('param')) is not int or s['param']!=0 for s in configs): + reasons.append('Every configured safety model must be noOutput with zero parameter') + pandas=snapshot.get('pandas') + if not isinstance(pandas,list) or not pandas: + reasons.append('Actual panda safety evidence is missing');pandas=[] + for index,panda in enumerate(pandas): + if not isinstance(panda,dict):reasons.append(f'Panda {index} evidence is malformed');continue + if panda.get('safetyModel')!='noOutput':reasons.append(f'Panda {index} is not actually noOutput') + if panda.get('controlsAllowed') is not False:reasons.append(f'Panda {index} permits or has unknown actuation') + if panda.get('faults')!=[]:reasons.append(f'Panda {index} fault state is not explicitly clear') + if panda.get('safetyRxChecksInvalid') is not False:reasons.append(f'Panda {index} safety receive checks are invalid or unknown') + for topic,maximum in [('pandaStates',2.),('carState',.5),('carControl',.5),('selfdriveState',.5)]: + if not _fresh(snapshot,topic,maximum):reasons.append(f'{topic} must be valid and fresh') + car=_mapping(snapshot.get('car')) + if car.get('canValid') is not True or car.get('canTimeout') is not False: + reasons.append('Actual car CAN input must be valid without timeout') + for name,fields in [('control',('enabled','latActive','longActive')),('selfdrive_state',('enabled','active'))]: + values=_mapping(snapshot.get(name)) + for field in fields: + if values.get(field) is not False:reasons.append(f'{name}.{field} must be explicitly inactive') + # Stable mode identity supplements, and never replaces, the complete car and + # baseline hashes. Runtime freshness/booleans are validated above, not hashed. + material={'mode':MODE,'passive':cp.get('passive'),'notCar':cp.get('notCar'), + 'dashcamOnly':cp.get('dashcamOnly'),'configured_safety':configs, + 'actual_panda_safety':[p.get('safetyModel') if isinstance(p,dict) else None for p in pandas]} + mode_identity=hashlib.sha256(json.dumps(material,sort_keys=True,separators=(',',':')).encode()).hexdigest() if not reasons else None + return {'mode':MODE,'safety_eligible':not reasons,'reasons':reasons,'mode_identity':mode_identity, + 'engagement_available':False,'label':'Passive observer — vehicle engagement unavailable'} + + +def authorization_mode_matches(record, assessment): + """Additional pin only. Does not grant user/coexistence authorization.""" + record=_mapping(record);assessment=_mapping(assessment) + return (assessment.get('safety_eligible') is True and record.get('live_mode')==MODE + and isinstance(assessment.get('mode_identity'),str) + and record.get('passive_mode_identity')==assessment['mode_identity']) diff --git a/roadscore/prototype/test_passive_observer_policy.py b/roadscore/prototype/test_passive_observer_policy.py new file mode 100644 index 0000000000..956a769f7f --- /dev/null +++ b/roadscore/prototype/test_passive_observer_policy.py @@ -0,0 +1,71 @@ +from copy import deepcopy +import pytest +from passive_observer_policy import MODE,evaluate_passive_observer,authorization_mode_matches + + +def healthy(): + return {'car_identity':{'passive':True,'notCar':False,'dashcamOnly':False,'safety':[{'model':'noOutput','param':0}]}, + 'pandas':[{'safetyModel':'noOutput','controlsAllowed':False,'faults':[],'safetyRxChecksInvalid':False}], + 'ages':{key:0.1 for key in ('pandaStates','carState','carControl','selfdriveState')}, + 'valid':{key:True for key in ('pandaStates','carState','carControl','selfdriveState')}, + 'car':{'canValid':True,'canTimeout':False}, + 'control':{'enabled':False,'latActive':False,'longActive':False}, + 'selfdrive_state':{'enabled':False,'active':False}} + + +def test_safety_eligible_is_not_authorization_or_ready(): + result=evaluate_passive_observer(healthy(),MODE) + assert result['safety_eligible'] and result['engagement_available'] is False + assert 'ready' not in result and 'coexistence_verified' not in result + assert not authorization_mode_matches({},result) + assert authorization_mode_matches({'live_mode':MODE,'passive_mode_identity':result['mode_identity']},result) + +@pytest.mark.parametrize('part,field,value',[ + ('car_identity','passive',False),('car_identity','passive',1),('car_identity','notCar',True), + ('car_identity','dashcamOnly',None),('control','enabled',True),('control','latActive',True), + ('control','longActive',True),('control','latActive',None),('selfdrive_state','enabled',True), + ('selfdrive_state','active',True),('car','canValid',False),('car','canTimeout',True)]) +def test_active_missing_or_invalid_vehicle_evidence_rejected(part,field,value): + snapshot=healthy();snapshot[part][field]=value + assert not evaluate_passive_observer(snapshot,MODE)['safety_eligible'] + +@pytest.mark.parametrize('field,value',[('safetyModel','tesla'),('safetyModel','silent'),('controlsAllowed',True), + ('controlsAllowed',None),('faults',['interruptRateCan2']),('faults',None),('safetyRxChecksInvalid',True)]) +def test_actual_panda_invariants(field,value): + snapshot=healthy();snapshot['pandas'][0][field]=value + result=evaluate_passive_observer(snapshot,MODE) + assert not result['safety_eligible'] and result['mode_identity'] is None + +@pytest.mark.parametrize('topic',('pandaStates','carState','carControl','selfdriveState')) +@pytest.mark.parametrize('age',(-.01,3.,float('nan'),float('inf'),None)) +def test_source_age_not_collector_liveness(topic,age): + snapshot=healthy();snapshot['ages'][topic]=age + assert not evaluate_passive_observer(snapshot,MODE)['safety_eligible'] + + +def test_toggle_or_offroad_nooutput_cannot_substitute_actual_passive_cp(): + snapshot=healthy();snapshot['car_identity']['passive']=False + snapshot['params']={'OpenpilotEnabledToggle':False,'AlwaysOnLateral':False};snapshot['offroad']=True + assert not evaluate_passive_observer(snapshot,MODE)['safety_eligible'] + + +def test_all_pandas_configs_and_authorization_pin(): + snapshot=healthy();baseline=evaluate_passive_observer(snapshot,MODE) + record={'live_mode':MODE,'passive_mode_identity':baseline['mode_identity']} + snapshot['pandas'].append(deepcopy(snapshot['pandas'][0])) + assert not authorization_mode_matches(record,evaluate_passive_observer(snapshot,MODE)) + snapshot['pandas'][1]['safetyModel']='tesla' + assert not evaluate_passive_observer(snapshot,MODE)['safety_eligible'] + snapshot=healthy();snapshot['car_identity']['safety'][0]['model']='tesla' + assert not evaluate_passive_observer(snapshot,MODE)['safety_eligible'] + + +def test_explicit_mode_no_auto_inference_and_input_unchanged(): + snapshot=healthy();before=deepcopy(snapshot) + assert not evaluate_passive_observer(snapshot,None)['safety_eligible'] + evaluate_passive_observer(snapshot,MODE);assert snapshot==before + +@pytest.mark.parametrize('field',('car_identity','ages','valid','car','control','selfdrive_state')) +def test_malformed_fields_fail_closed(field): + snapshot=healthy();snapshot[field]=None + assert not evaluate_passive_observer(snapshot,MODE)['safety_eligible']