From 06b5a3d6d13a66322555d67b7db11b40e9f40f59 Mon Sep 17 00:00:00 2001 From: virtuscyber Date: Sat, 11 Apr 2026 13:47:08 -0400 Subject: [PATCH] feat(obd-ai): add policy guardrails and audit logging --- docs/LLM Tool Server.md | 3 + obd_ai/__init__.py | 28 ++ obd_ai/audit.py | 91 ++++++ obd_ai/catalog.py | 9 + obd_ai/policy.py | 272 +++++++++++++++++ obd_ai/serializers.py | 3 + obd_ai/session.py | 44 ++- obd_ai/tools.py | 550 ++++++++++++++++++++++++++++------- tests/test_obd_ai_catalog.py | 3 + tests/test_obd_ai_policy.py | 53 ++++ tests/test_obd_ai_server.py | 11 + tests/test_obd_ai_session.py | 21 +- tests/test_obd_ai_tools.py | 73 ++++- 13 files changed, 1048 insertions(+), 113 deletions(-) create mode 100644 obd_ai/audit.py create mode 100644 obd_ai/policy.py create mode 100644 tests/test_obd_ai_policy.py diff --git a/docs/LLM Tool Server.md b/docs/LLM Tool Server.md index 68f16499..3edcb33b 100644 --- a/docs/LLM Tool Server.md +++ b/docs/LLM Tool Server.md @@ -17,6 +17,7 @@ This transport is function-calling-friendly JSON over HTTP and reuses the same c - `basic_health_scan` These tools remain read-only. The server does **not** expose raw OBD command construction or write operations. +Sensitive or destructive actions such as `CLEAR_DTC` are covered by the policy layer, which returns structured approval-required or blocked responses instead of executing them. ## Start the server @@ -81,4 +82,6 @@ Equivalent convenience route for directly calling a specific tool. - Session lifecycle is managed by the underlying tool surface. - Call `connect_vehicle` before issuing read operations. +- The tool surface emits structured audit events for tool invocations and policy decisions. +- Future motion, ignition, and vehicle-state guards plug into the policy metadata without exposing raw commands. - Error payloads are returned in structured JSON and preserve existing `obd_ai` error codes when possible. diff --git a/obd_ai/__init__.py b/obd_ai/__init__.py index ddfd6013..0d7046e2 100644 --- a/obd_ai/__init__.py +++ b/obd_ai/__init__.py @@ -1,10 +1,24 @@ """Safe, high-level interfaces for LLM/MCP-style OBD access.""" +from .audit import AuditEvent, InMemoryAuditSink, LoggerAuditSink from .catalog import ( ApprovedCommand, ApprovedCommandCatalog, DEFAULT_APPROVED_COMMAND_CATALOG, ) +from .policy import ( + DEFAULT_POLICY_CATALOG, + DECISION_ALLOW, + DECISION_DENY, + DECISION_REQUIRE_APPROVAL, + OBDAIPolicyEngine, + POLICY_CLASS_BLOCKED, + POLICY_CLASS_DESTRUCTIVE, + POLICY_CLASS_READ_ONLY, + POLICY_CLASS_SENSITIVE, + PolicyAction, + PolicyDecision, +) from .serializers import ( serialize_approved_command, serialize_approved_command_catalog, @@ -19,10 +33,24 @@ __all__ = [ "ApprovedCommand", "ApprovedCommandCatalog", + "AuditEvent", "DEFAULT_APPROVED_COMMAND_CATALOG", + "DEFAULT_POLICY_CATALOG", + "DECISION_ALLOW", + "DECISION_DENY", + "DECISION_REQUIRE_APPROVAL", + "InMemoryAuditSink", + "LoggerAuditSink", + "OBDAIPolicyEngine", "OBDAISession", "OBDAISessionManager", "OBDAIReadOnlyToolSurface", + "POLICY_CLASS_BLOCKED", + "POLICY_CLASS_DESTRUCTIVE", + "POLICY_CLASS_READ_ONLY", + "POLICY_CLASS_SENSITIVE", + "PolicyAction", + "PolicyDecision", "serialize_approved_command", "serialize_approved_command_catalog", "serialize_connection_metadata", diff --git a/obd_ai/audit.py b/obd_ai/audit.py new file mode 100644 index 00000000..03a13445 --- /dev/null +++ b/obd_ai/audit.py @@ -0,0 +1,91 @@ +"""Audit logging primitives for the OBD AI tool surface.""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, Mapping, Optional, Protocol + + +@dataclass(frozen=True) +class AuditEvent: + """Structured audit event emitted by the OBD AI layer.""" + + event_type: str + tool: str + requested_action: str + request: Dict[str, Any] + decision: Optional[str] = None + result: Optional[str] = None + error_code: Optional[str] = None + policy: Optional[Dict[str, Any]] = None + metadata: Dict[str, Any] = field(default_factory=dict) + occurred_at: str = field( + default_factory=lambda: datetime.now(timezone.utc).isoformat() + ) + + def to_dict(self) -> Dict[str, Any]: + payload = { + "event_type": self.event_type, + "tool": self.tool, + "requested_action": self.requested_action, + "request": dict(self.request), + "decision": self.decision, + "result": self.result, + "error_code": self.error_code, + "policy": dict(self.policy) if self.policy is not None else None, + "metadata": dict(self.metadata), + "occurred_at": self.occurred_at, + } + return payload + + +class AuditSink(Protocol): + def record(self, event: AuditEvent) -> None: + """Persist or forward an audit event.""" + + +class LoggerAuditSink: + """Default audit sink that writes structured events to ``logging``.""" + + def __init__(self, logger: Optional[logging.Logger] = None): + self._logger = logger or logging.getLogger("obd_ai.audit") + + def record(self, event: AuditEvent) -> None: + self._logger.info("obd_ai_audit %s", json.dumps(event.to_dict(), sort_keys=True)) + + +class InMemoryAuditSink: + """Test-friendly audit sink that stores serialized events in memory.""" + + def __init__(self): + self.events = [] + + def record(self, event: AuditEvent) -> None: + self.events.append(event.to_dict()) + + +def build_audit_event( + event_type: str, + tool: str, + requested_action: str, + request: Optional[Mapping[str, Any]] = None, + decision: Optional[str] = None, + result: Optional[str] = None, + error_code: Optional[str] = None, + policy: Optional[Mapping[str, Any]] = None, + metadata: Optional[Mapping[str, Any]] = None, +) -> AuditEvent: + return AuditEvent( + event_type=event_type, + tool=tool, + requested_action=requested_action, + request=dict(request or {}), + decision=decision, + result=result, + error_code=error_code, + policy=dict(policy) if policy is not None else None, + metadata=dict(metadata or {}), + ) diff --git a/obd_ai/catalog.py b/obd_ai/catalog.py index f7792a6c..3ae250d7 100644 --- a/obd_ai/catalog.py +++ b/obd_ai/catalog.py @@ -18,6 +18,9 @@ class ApprovedCommand: public_name: str category: str description: str + policy_class: str + approval_required: bool + guard_tags: Tuple[str, ...] obd_command_name: str obd_command: OBDCommand @@ -29,6 +32,9 @@ class _CommandDefinition: category: str description: str obd_command_name: str + policy_class: str = "read_only" + approval_required: bool = False + guard_tags: Tuple[str, ...] = () class ApprovedCommandCatalog: @@ -54,6 +60,9 @@ def __init__(self, definitions: Iterable[_CommandDefinition]): public_name=definition.public_name, category=definition.category, description=definition.description, + policy_class=definition.policy_class, + approval_required=definition.approval_required, + guard_tags=definition.guard_tags, obd_command_name=definition.obd_command_name, obd_command=obd_command, ) diff --git a/obd_ai/policy.py b/obd_ai/policy.py new file mode 100644 index 00000000..93e92b92 --- /dev/null +++ b/obd_ai/policy.py @@ -0,0 +1,272 @@ +"""Policy guardrails for the OBD AI tool surface.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from types import MappingProxyType +from typing import Any, Dict, Iterable, Mapping, Optional, Tuple + +from .catalog import ApprovedCommandCatalog, DEFAULT_APPROVED_COMMAND_CATALOG + +POLICY_CLASS_READ_ONLY = "read_only" +POLICY_CLASS_SENSITIVE = "sensitive" +POLICY_CLASS_DESTRUCTIVE = "destructive" +POLICY_CLASS_BLOCKED = "blocked" + +DECISION_ALLOW = "allow" +DECISION_REQUIRE_APPROVAL = "require_approval" +DECISION_DENY = "deny" + + +@dataclass(frozen=True) +class PolicyAction: + """Policy metadata for an action the OBD AI layer may encounter.""" + + key: str + public_name: str + description: str + policy_class: str + obd_command_name: Optional[str] = None + default_decision: str = DECISION_ALLOW + llm_exposed: bool = True + approval_required: bool = False + guard_tags: Tuple[str, ...] = () + reason: Optional[str] = None + + +@dataclass(frozen=True) +class PolicyDecision: + """Resolved policy decision for a requested action.""" + + action: PolicyAction + decision: str + reason: str + details: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "action_key": self.action.key, + "public_name": self.action.public_name, + "policy_class": self.action.policy_class, + "decision": self.decision, + "reason": self.reason, + "approval_required": self.action.approval_required, + "llm_exposed": self.action.llm_exposed, + "obd_command_name": self.action.obd_command_name, + "guard_tags": list(self.action.guard_tags), + "details": dict(self.details), + } + + +class PolicyCatalog: + """Read-only registry of policy definitions.""" + + def __init__(self, actions: Iterable[PolicyAction]): + mutable_by_key = {} + mutable_by_obd_command_name = {} + ordered = [] + + for action in actions: + if action.key in mutable_by_key: + raise ValueError(f"Duplicate policy action key: {action.key}") + + mutable_by_key[action.key] = action + ordered.append(action) + + if action.obd_command_name is not None: + mutable_by_obd_command_name[action.obd_command_name.upper()] = action + + self._by_key = MappingProxyType(mutable_by_key) + self._by_obd_command_name = MappingProxyType(mutable_by_obd_command_name) + self._ordered = tuple(ordered) + + def get(self, key: str) -> PolicyAction: + try: + return self._by_key[key] + except KeyError as exc: + raise KeyError(f"Unknown policy action key: {key}") from exc + + def list(self) -> Tuple[PolicyAction, ...]: + return self._ordered + + def find_action(self, action_key: str) -> Optional[PolicyAction]: + action = self._by_key.get(action_key) + if action is not None: + return action + + normalized = action_key.upper() + return self._by_obd_command_name.get(normalized) + + +class OBDAIPolicyEngine: + """Evaluate tool requests against the OBD AI guardrail policy catalog.""" + + _RAW_COMMAND_FIELDS = ( + "raw_command", + "custom_command", + "obd_command", + "obd_command_name", + "mode", + "pid", + "bytes", + "header", + ) + + def __init__(self, catalog: PolicyCatalog): + self._catalog = catalog + + @classmethod + def from_approved_catalog( + cls, + approved_catalog: ApprovedCommandCatalog = DEFAULT_APPROVED_COMMAND_CATALOG, + ) -> "OBDAIPolicyEngine": + return cls(build_default_policy_catalog(approved_catalog)) + + def evaluate_request( + self, + *, + command_key: Optional[str], + payload: Optional[Mapping[str, Any]] = None, + ) -> Optional[PolicyDecision]: + request_payload = dict(payload or {}) + + if request_payload.get("force"): + action = self._catalog.get("force_query") + return PolicyDecision( + action=action, + decision=DECISION_DENY, + reason="High-level OBD AI tools do not allow force=True bypasses.", + details={"requested_force": request_payload.get("force")}, + ) + + blocked_fields = [ + field_name + for field_name in self._RAW_COMMAND_FIELDS + if field_name in request_payload + ] + if blocked_fields: + action = self._catalog.get("custom_command") + return PolicyDecision( + action=action, + decision=DECISION_DENY, + reason="Raw or custom OBD command execution is blocked in the MVP LLM surface.", + details={"blocked_fields": blocked_fields}, + ) + + if command_key is None: + return None + + action = self._catalog.find_action(command_key) + if action is None: + action = self._catalog.get("unknown_command") + return PolicyDecision( + action=action, + decision=DECISION_DENY, + reason="Requested command is not defined in the approved policy catalog.", + details={"command_key": command_key}, + ) + + if action.default_decision == DECISION_ALLOW: + return PolicyDecision( + action=action, + decision=DECISION_ALLOW, + reason=action.reason or "Requested action is approved for read-only access.", + ) + + if action.default_decision == DECISION_REQUIRE_APPROVAL: + return PolicyDecision( + action=action, + decision=DECISION_REQUIRE_APPROVAL, + reason=action.reason or "Requested action requires explicit approval before execution.", + details={ + "approval": { + "required": True, + "status": "not_implemented", + "integration": "approval_hook_placeholder", + } + }, + ) + + return PolicyDecision( + action=action, + decision=DECISION_DENY, + reason=action.reason or "Requested action is blocked by policy.", + ) + + +def build_default_policy_catalog( + approved_catalog: ApprovedCommandCatalog = DEFAULT_APPROVED_COMMAND_CATALOG, +) -> PolicyCatalog: + actions = [ + PolicyAction( + key=approved.key, + public_name=approved.public_name, + description=approved.description, + policy_class=approved.policy_class, + obd_command_name=approved.obd_command_name, + default_decision=DECISION_ALLOW, + llm_exposed=True, + approval_required=approved.approval_required, + guard_tags=approved.guard_tags, + reason="Approved read-only diagnostic command.", + ) + for approved in approved_catalog.list() + ] + + actions.extend( + [ + PolicyAction( + key="clear_trouble_codes", + public_name="Clear Trouble Codes", + description="Clear stored DTCs and freeze-frame data.", + policy_class=POLICY_CLASS_DESTRUCTIVE, + obd_command_name="CLEAR_DTC", + default_decision=DECISION_REQUIRE_APPROVAL, + llm_exposed=False, + approval_required=True, + guard_tags=( + "vehicle_stationary", + "ignition_state", + "vehicle_state", + "operator_approval", + ), + reason="CLEAR_DTC is destructive and must flow through an explicit approval hook.", + ), + PolicyAction( + key="custom_command", + public_name="Custom OBD Command", + description="Arbitrary raw/custom OBD command execution.", + policy_class=POLICY_CLASS_BLOCKED, + default_decision=DECISION_DENY, + llm_exposed=False, + approval_required=False, + reason="Arbitrary custom commands are blocked from the MVP LLM surface.", + ), + PolicyAction( + key="force_query", + public_name="Force Query", + description="Bypass command support checks with force=True.", + policy_class=POLICY_CLASS_BLOCKED, + default_decision=DECISION_DENY, + llm_exposed=False, + approval_required=False, + guard_tags=("vehicle_state",), + reason="Force-query bypasses are blocked from the high-level OBD AI interface.", + ), + PolicyAction( + key="unknown_command", + public_name="Unknown Command", + description="Action not present in the approved OBD AI policy catalog.", + policy_class=POLICY_CLASS_BLOCKED, + default_decision=DECISION_DENY, + llm_exposed=False, + approval_required=False, + reason="Only explicitly cataloged commands may reach the OBD AI surface.", + ), + ] + ) + + return PolicyCatalog(actions) + + +DEFAULT_POLICY_CATALOG = build_default_policy_catalog() diff --git a/obd_ai/serializers.py b/obd_ai/serializers.py index efff11b8..bdcbf887 100644 --- a/obd_ai/serializers.py +++ b/obd_ai/serializers.py @@ -121,6 +121,9 @@ def serialize_approved_command( "public_name": approved.public_name, "category": approved.category, "description": approved.description, + "policy_class": approved.policy_class, + "approval_required": approved.approval_required, + "guard_tags": list(approved.guard_tags), "obd_command_name": approved.obd_command_name, "obd_command": serialize_obd_command(approved.obd_command), } diff --git a/obd_ai/session.py b/obd_ai/session.py index ab601b1e..fbaf900d 100644 --- a/obd_ai/session.py +++ b/obd_ai/session.py @@ -9,6 +9,7 @@ from obd.utils import OBDStatus from .catalog import ApprovedCommandCatalog, DEFAULT_APPROVED_COMMAND_CATALOG +from .policy import DECISION_ALLOW, OBDAIPolicyEngine from .serializers import ( serialize_approved_command, serialize_connection_metadata, @@ -21,9 +22,17 @@ class OBDAISession: """Safe wrapper around an OBD connection constrained by an approved catalog.""" - def __init__(self, connection: obd.OBD, catalog: ApprovedCommandCatalog): + def __init__( + self, + connection: obd.OBD, + catalog: ApprovedCommandCatalog, + policy_engine: Optional[OBDAIPolicyEngine] = None, + ): self._connection = connection self._catalog = catalog + self._policy_engine = policy_engine or OBDAIPolicyEngine.from_approved_catalog( + catalog + ) def __enter__(self) -> "OBDAISession": return self @@ -45,12 +54,23 @@ def list_approved_command_descriptors(self, include_support: bool = False): ) return descriptors - def query(self, approved_key: str) -> OBDResponse: + def query(self, approved_key: str, *, force: bool = False) -> OBDResponse: + decision = self._policy_engine.evaluate_request( + command_key=approved_key, + payload={"command_key": approved_key, "force": force}, + ) + if ( + decision is not None + and decision.decision != DECISION_ALLOW + and decision.action.key != "unknown_command" + ): + raise PermissionError(decision.reason) + approved = self._catalog.get(approved_key) return self._connection.query(approved.obd_command) - def query_serialized(self, approved_key: str): - return serialize_obd_response(self.query(approved_key)) + def query_serialized(self, approved_key: str, *, force: bool = False): + return serialize_obd_response(self.query(approved_key, force=force)) def supports(self, approved_key: str) -> bool: approved = self._catalog.get(approved_key) @@ -102,9 +122,13 @@ def __init__( self, catalog: ApprovedCommandCatalog = DEFAULT_APPROVED_COMMAND_CATALOG, obd_factory: Optional[OBDFactory] = None, + policy_engine: Optional[OBDAIPolicyEngine] = None, ): self._catalog = catalog self._obd_factory = obd_factory or obd.OBD + self._policy_engine = policy_engine or OBDAIPolicyEngine.from_approved_catalog( + catalog + ) def open_session( self, @@ -125,7 +149,15 @@ def open_session( check_voltage=check_voltage, start_low_power=start_low_power, ) - return OBDAISession(connection=connection, catalog=self._catalog) + return OBDAISession( + connection=connection, + catalog=self._catalog, + policy_engine=self._policy_engine, + ) def wrap_existing(self, connection: obd.OBD) -> OBDAISession: - return OBDAISession(connection=connection, catalog=self._catalog) + return OBDAISession( + connection=connection, + catalog=self._catalog, + policy_engine=self._policy_engine, + ) diff --git a/obd_ai/tools.py b/obd_ai/tools.py index dab8cf44..5d8cef12 100644 --- a/obd_ai/tools.py +++ b/obd_ai/tools.py @@ -2,9 +2,16 @@ from __future__ import annotations -from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple +from typing import Any, Dict, List, Mapping, Optional, Sequence, Set, Tuple +from .audit import AuditSink, LoggerAuditSink, build_audit_event from .catalog import ApprovedCommandCatalog, DEFAULT_APPROVED_COMMAND_CATALOG +from .policy import ( + DECISION_ALLOW, + DECISION_DENY, + DECISION_REQUIRE_APPROVAL, + OBDAIPolicyEngine, +) from .serializers import serialize_approved_command_catalog, serialize_obd_response from .session import OBDAISession, OBDAISessionManager @@ -35,12 +42,24 @@ def __init__( self, session_manager: Optional[OBDAISessionManager] = None, catalog: Optional[ApprovedCommandCatalog] = None, + policy_engine: Optional[OBDAIPolicyEngine] = None, + audit_sink: Optional[AuditSink] = None, ): resolved_catalog = catalog or DEFAULT_APPROVED_COMMAND_CATALOG + resolved_policy_engine = policy_engine or OBDAIPolicyEngine.from_approved_catalog( + resolved_catalog + ) self._session_manager = session_manager or OBDAISessionManager( - catalog=resolved_catalog + catalog=resolved_catalog, + policy_engine=resolved_policy_engine, ) self._catalog = getattr(self._session_manager, "_catalog", resolved_catalog) + self._policy_engine = getattr( + self._session_manager, + "_policy_engine", + resolved_policy_engine, + ) + self._audit_sink = audit_sink or LoggerAuditSink() self._session: Optional[OBDAISession] = None def connect_vehicle(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: @@ -59,11 +78,14 @@ def connect_vehicle(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str unknown_fields = sorted(set(payload.keys()) - allowed_fields) if unknown_fields: - return self._error( - tool="connect_vehicle", - code="invalid_input", - message="Unsupported connect_vehicle input fields.", - details={"unknown_fields": unknown_fields}, + return self._audit_and_return( + self._error( + tool="connect_vehicle", + code="invalid_input", + message="Unsupported connect_vehicle input fields.", + details={"unknown_fields": unknown_fields}, + ), + request_payload=payload, ) self.close() @@ -71,26 +93,39 @@ def connect_vehicle(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str metadata = self._session.connection_metadata() if not metadata["is_connected"]: - return self._error( - tool="connect_vehicle", - code="disconnected_adapter", - message="OBD adapter is not connected to a vehicle.", - details={"connection": metadata}, + return self._audit_and_return( + self._error( + tool="connect_vehicle", + code="disconnected_adapter", + message="OBD adapter is not connected to a vehicle.", + details={"connection": metadata}, + ), + request_payload=payload, ) - return self._success("connect_vehicle", {"connection": metadata}) + return self._audit_and_return( + self._success("connect_vehicle", {"connection": metadata}), + request_payload=payload, + ) def get_vehicle_status(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: """Return transport-level vehicle connection status.""" - del input + payload = dict(input or {}) + input_error = self._validate_no_extra_input("get_vehicle_status", payload) + if input_error is not None: + return input_error + session, error = self._require_connected(tool="get_vehicle_status") if error is not None: - return error - - return self._success( - "get_vehicle_status", - {"connection": session.connection_metadata()}, + return self._audit_and_return(error, request_payload=payload) + + return self._audit_and_return( + self._success( + "get_vehicle_status", + {"connection": session.connection_metadata()}, + ), + request_payload=payload, ) def list_supported_commands( @@ -100,20 +135,31 @@ def list_supported_commands( """List approved commands, optionally including adapter support status.""" payload = dict(input or {}) + input_error = self._validate_allowed_fields( + "list_supported_commands", + payload, + {"include_support", "only_supported"}, + ) + if input_error is not None: + return input_error + include_support = bool(payload.get("include_support", True)) only_supported = bool(payload.get("only_supported", False)) if only_supported and not include_support: - return self._error( - tool="list_supported_commands", - code="invalid_input", - message="only_supported=true requires include_support=true.", + return self._audit_and_return( + self._error( + tool="list_supported_commands", + code="invalid_input", + message="only_supported=true requires include_support=true.", + ), + request_payload=payload, ) if include_support: session, error = self._require_connected(tool="list_supported_commands") if error is not None: - return error + return self._audit_and_return(error, request_payload=payload) commands = session.list_approved_command_descriptors(include_support=True) else: commands = serialize_approved_command_catalog(self._catalog) @@ -121,54 +167,114 @@ def list_supported_commands( if only_supported: commands = [command for command in commands if command.get("supported")] - return self._success( - "list_supported_commands", - { - "commands": commands, - "count": len(commands), - }, + return self._audit_and_return( + self._success( + "list_supported_commands", + { + "commands": commands, + "count": len(commands), + }, + ), + request_payload=payload, ) def read_sensor(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: """Read one approved sensor key.""" payload = dict(input or {}) + input_error = self._validate_allowed_fields( + "read_sensor", + payload, + { + "command_key", + "force", + "raw_command", + "custom_command", + "obd_command", + "obd_command_name", + "mode", + "pid", + "bytes", + "header", + }, + ) + if input_error is not None: + return input_error + command_key = payload.get("command_key") if not isinstance(command_key, str) or not command_key: - return self._error( - tool="read_sensor", - code="invalid_input", - message="read_sensor requires a non-empty string command_key.", + return self._audit_and_return( + self._error( + tool="read_sensor", + code="invalid_input", + message="read_sensor requires a non-empty string command_key.", + ), + request_payload=payload, ) response, error = self._query_approved_command( tool="read_sensor", command_key=command_key, + request_payload=payload, ) if error is not None: - return error + return self._audit_and_return(error, request_payload=payload) - return self._success( - "read_sensor", - { - "command_key": command_key, - "response": response, - }, + return self._audit_and_return( + self._success( + "read_sensor", + { + "command_key": command_key, + "response": response, + }, + ), + request_payload=payload, ) def read_sensors(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: """Read multiple approved sensor keys with per-key status.""" payload = dict(input or {}) + input_error = self._validate_allowed_fields( + "read_sensors", + payload, + {"command_keys", "force"}, + ) + if input_error is not None: + return input_error + command_keys = payload.get("command_keys") if not isinstance(command_keys, Sequence) or isinstance(command_keys, (str, bytes)): - return self._error( + return self._audit_and_return( + self._error( + tool="read_sensors", + code="invalid_input", + message="read_sensors requires command_keys as a list of strings.", + ), + request_payload=payload, + ) + + batch_decision = self._policy_engine.evaluate_request( + command_key=None, + payload=payload, + ) + if batch_decision is not None and batch_decision.decision != DECISION_ALLOW: + error = self._policy_error_payload( tool="read_sensors", - code="invalid_input", - message="read_sensors requires command_keys as a list of strings.", + decision=batch_decision, + command_key=None, ) + self._record_policy_event( + tool="read_sensors", + command_key=batch_decision.action.key, + request_payload=payload, + decision=batch_decision, + result="blocked", + error_code=error["error"]["code"], + ) + return self._audit_and_return(error, request_payload=payload) results = [] has_errors = False @@ -192,6 +298,7 @@ def read_sensors(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, A response, error = self._query_approved_command( tool="read_sensors", command_key=command_key, + request_payload={"command_key": command_key, **payload}, ) if error is not None: has_errors = True @@ -212,44 +319,62 @@ def read_sensors(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, A } ) - return { - "ok": not has_errors, - "tool": "read_sensors", - "data": { - "results": results, - "count": len(results), + return self._audit_and_return( + { + "ok": not has_errors, + "tool": "read_sensors", + "data": { + "results": results, + "count": len(results), + }, }, - } + request_payload=payload, + ) def get_dtc_codes(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: """Read stored and pending DTC code lists.""" - del input - return self._multi_read( - tool="get_dtc_codes", - reads={ - "stored": "stored_trouble_codes", - "pending": "pending_trouble_codes", - }, + payload = dict(input or {}) + input_error = self._validate_no_extra_input("get_dtc_codes", payload) + if input_error is not None: + return input_error + + return self._audit_and_return( + self._multi_read( + tool="get_dtc_codes", + reads={ + "stored": "stored_trouble_codes", + "pending": "pending_trouble_codes", + }, + ), + request_payload=payload, ) def get_freeze_frame(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: """Read freeze-frame trigger code (DTC).""" - del input + payload = dict(input or {}) + input_error = self._validate_no_extra_input("get_freeze_frame", payload) + if input_error is not None: + return input_error + response, error = self._query_approved_command( tool="get_freeze_frame", command_key="freeze_frame_trouble_code", + request_payload=payload, ) if error is not None: - return error + return self._audit_and_return(error, request_payload=payload) - return self._success( - "get_freeze_frame", - { - "command_key": "freeze_frame_trouble_code", - "response": response, - }, + return self._audit_and_return( + self._success( + "get_freeze_frame", + { + "command_key": "freeze_frame_trouble_code", + "response": response, + }, + ), + request_payload=payload, ) def get_emissions_monitor_status( @@ -258,29 +383,44 @@ def get_emissions_monitor_status( ) -> Dict[str, Any]: """Read emissions monitor status (mode 01 PID 01).""" - del input + payload = dict(input or {}) + input_error = self._validate_no_extra_input( + "get_emissions_monitor_status", + payload, + ) + if input_error is not None: + return input_error + response, error = self._query_approved_command( tool="get_emissions_monitor_status", command_key="status_since_dtc_clear", + request_payload=payload, ) if error is not None: - return error + return self._audit_and_return(error, request_payload=payload) - return self._success( - "get_emissions_monitor_status", - { - "command_key": "status_since_dtc_clear", - "response": response, - }, + return self._audit_and_return( + self._success( + "get_emissions_monitor_status", + { + "command_key": "status_since_dtc_clear", + "response": response, + }, + ), + request_payload=payload, ) def basic_health_scan(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: """Run a deterministic baseline read-only health scan.""" - del input + payload = dict(input or {}) + input_error = self._validate_no_extra_input("basic_health_scan", payload) + if input_error is not None: + return input_error + session, error = self._require_connected(tool="basic_health_scan") if error is not None: - return error + return self._audit_and_return(error, request_payload=payload) status = self._scan_read("status_since_dtc_clear") dtc_snapshot = { @@ -299,22 +439,25 @@ def basic_health_scan(self, input: Optional[Mapping[str, Any]] = None) -> Dict[s live_metrics=live_metrics, ) - return self._success( - "basic_health_scan", - { - "scan_version": "basic_health_scan.v1", - "connection": session.connection_metadata(), - "snapshot": { - "status": status, - "dtc": dtc_snapshot, - "readiness": status, - "live_metrics": live_metrics, + return self._audit_and_return( + self._success( + "basic_health_scan", + { + "scan_version": "basic_health_scan.v1", + "connection": session.connection_metadata(), + "snapshot": { + "status": status, + "dtc": dtc_snapshot, + "readiness": status, + "live_metrics": live_metrics, + }, + "findings": findings, + "anomalies": anomalies, + "next_steps": next_steps, + "summary": summary, }, - "findings": findings, - "anomalies": anomalies, - "next_steps": next_steps, - "summary": summary, - }, + ), + request_payload=payload, ) def close(self) -> None: @@ -329,7 +472,11 @@ def _multi_read(self, tool: str, reads: Mapping[str, str]) -> Dict[str, Any]: has_errors = False for label, command_key in reads.items(): - response, error = self._query_approved_command(tool=tool, command_key=command_key) + response, error = self._query_approved_command( + tool=tool, + command_key=command_key, + request_payload={"command_key": command_key}, + ) if error is not None: has_errors = True results[label] = { @@ -354,6 +501,7 @@ def _scan_read(self, command_key: str) -> Dict[str, Any]: response, error = self._query_approved_command( tool="basic_health_scan", command_key=command_key, + request_payload={"command_key": command_key}, ) if error is not None: return { @@ -669,45 +817,137 @@ def _query_approved_command( self, tool: str, command_key: str, + request_payload: Optional[Mapping[str, Any]] = None, ) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]: - if command_key not in self._catalog: - return None, self._error( + payload = dict(request_payload or {"command_key": command_key}) + decision = self._policy_engine.evaluate_request( + command_key=command_key, + payload=payload, + ) + if decision is not None and decision.decision != DECISION_ALLOW: + error = self._policy_error_payload( tool=tool, - code="unsupported_pid", - message="Requested command is not in the approved read-only catalog.", - details={"command_key": command_key}, + decision=decision, + command_key=command_key, + ) + self._record_policy_event( + tool=tool, + command_key=command_key, + request_payload=payload, + decision=decision, + result="blocked", + error_code=error["error"]["code"], ) + return None, error session, error = self._require_connected(tool=tool) if error is not None: + if decision is not None: + self._record_policy_event( + tool=tool, + command_key=command_key, + request_payload=payload, + decision=decision, + result="disconnected", + error_code=error["error"]["code"], + ) return None, error + if command_key not in self._catalog: + unsupported_error = self._error( + tool=tool, + code="unsupported_pid", + message="Requested command is not in the approved read-only catalog.", + details={ + "command_key": command_key, + "policy": decision.to_dict() if decision is not None else None, + }, + ) + if decision is not None: + self._record_policy_event( + tool=tool, + command_key=command_key, + request_payload=payload, + decision=decision, + result="blocked", + error_code=unsupported_error["error"]["code"], + ) + return None, unsupported_error + if not session.supports(command_key): - return None, self._error( + unsupported_error = self._error( tool=tool, code="unsupported_pid", message="Requested PID is not supported by the connected vehicle.", - details={"command_key": command_key}, + details={ + "command_key": command_key, + "policy": decision.to_dict() if decision is not None else None, + }, ) + if decision is not None: + self._record_policy_event( + tool=tool, + command_key=command_key, + request_payload=payload, + decision=decision, + result="vehicle_unsupported", + error_code=unsupported_error["error"]["code"], + ) + return None, unsupported_error try: response = session.query(command_key) except Exception as exc: # pragma: no branch if self._is_timeout_error(exc): - return None, self._error( + timeout_error = self._error( tool=tool, code="timeout", message="Timed out waiting for a response from the adapter.", - details={"command_key": command_key}, + details={ + "command_key": command_key, + "policy": decision.to_dict() if decision is not None else None, + }, ) + if decision is not None: + self._record_policy_event( + tool=tool, + command_key=command_key, + request_payload=payload, + decision=decision, + result="timeout", + error_code=timeout_error["error"]["code"], + ) + return None, timeout_error raise if response.is_null(): - return None, self._error( + null_error = self._error( tool=tool, code="null_response", message="Adapter returned no data for the requested PID.", - details={"command_key": command_key}, + details={ + "command_key": command_key, + "policy": decision.to_dict() if decision is not None else None, + }, + ) + if decision is not None: + self._record_policy_event( + tool=tool, + command_key=command_key, + request_payload=payload, + decision=decision, + result="null_response", + error_code=null_error["error"]["code"], + ) + return None, null_error + + if decision is not None: + self._record_policy_event( + tool=tool, + command_key=command_key, + request_payload=payload, + decision=decision, + result="queried", ) return serialize_obd_response(response), None @@ -733,6 +973,110 @@ def _require_connected( return self._session, None + def _validate_no_extra_input( + self, + tool: str, + payload: Mapping[str, Any], + ) -> Optional[Dict[str, Any]]: + return self._validate_allowed_fields(tool, payload, set()) + + def _validate_allowed_fields( + self, + tool: str, + payload: Mapping[str, Any], + allowed_fields: Set[str], + ) -> Optional[Dict[str, Any]]: + unknown_fields = sorted(set(payload.keys()) - set(allowed_fields)) + if not unknown_fields: + return None + + return self._audit_and_return( + self._error( + tool=tool, + code="invalid_input", + message=f"Unsupported {tool} input fields.", + details={"unknown_fields": unknown_fields}, + ), + request_payload=payload, + ) + + def _policy_error_payload( + self, + tool: str, + decision: Any, + command_key: Optional[str], + ) -> Dict[str, Any]: + if decision.decision == DECISION_REQUIRE_APPROVAL: + code = "approval_required" + elif decision.action.key in {"custom_command", "force_query"}: + code = "blocked_action" + else: + code = "unsupported_pid" + + details = dict(decision.to_dict()) + if command_key is not None: + details.setdefault("command_key", command_key) + + return self._error( + tool=tool, + code=code, + message=decision.reason, + details=details, + ) + + def _audit_and_return( + self, + payload: Dict[str, Any], + request_payload: Optional[Mapping[str, Any]] = None, + ) -> Dict[str, Any]: + self._record_tool_event( + tool=payload.get("tool", "unknown"), + request_payload=request_payload, + result_payload=payload, + ) + return payload + + def _record_tool_event( + self, + tool: str, + request_payload: Optional[Mapping[str, Any]], + result_payload: Mapping[str, Any], + ) -> None: + self._audit_sink.record( + build_audit_event( + event_type="tool_invocation", + tool=tool, + requested_action=tool, + request=request_payload, + decision="success" if result_payload.get("ok") else "error", + result="success" if result_payload.get("ok") else "error", + error_code=result_payload.get("error", {}).get("code"), + metadata={"response_ok": bool(result_payload.get("ok"))}, + ) + ) + + def _record_policy_event( + self, + tool: str, + command_key: str, + request_payload: Optional[Mapping[str, Any]], + decision: Any, + result: str, + error_code: Optional[str] = None, + ) -> None: + self._audit_sink.record( + build_audit_event( + event_type="policy_decision", + tool=tool, + requested_action=command_key, + request=request_payload, + decision=decision.decision, + result=result, + error_code=error_code, + policy=decision.to_dict(), + ) + ) + @staticmethod def _success(tool: str, data: Mapping[str, Any]) -> Dict[str, Any]: return { diff --git a/tests/test_obd_ai_catalog.py b/tests/test_obd_ai_catalog.py index 2d8c977d..ecc6e666 100644 --- a/tests/test_obd_ai_catalog.py +++ b/tests/test_obd_ai_catalog.py @@ -10,6 +10,9 @@ def test_default_catalog_maps_to_real_obd_commands(): rpm = catalog.get("engine_rpm") assert rpm.public_name == "Engine RPM" assert rpm.category == "powertrain" + assert rpm.policy_class == "read_only" + assert rpm.approval_required is False + assert rpm.guard_tags == () assert rpm.obd_command_name == "RPM" assert rpm.obd_command is obd.commands.RPM diff --git a/tests/test_obd_ai_policy.py b/tests/test_obd_ai_policy.py new file mode 100644 index 00000000..2597190f --- /dev/null +++ b/tests/test_obd_ai_policy.py @@ -0,0 +1,53 @@ +from obd_ai.policy import ( + DECISION_ALLOW, + DECISION_DENY, + DECISION_REQUIRE_APPROVAL, + OBDAIPolicyEngine, +) + + +policy = OBDAIPolicyEngine.from_approved_catalog() + + +def test_policy_engine_allows_approved_read_only_command(): + decision = policy.evaluate_request( + command_key="engine_rpm", + payload={"command_key": "engine_rpm"}, + ) + + assert decision is not None + assert decision.decision == DECISION_ALLOW + assert decision.to_dict()["policy_class"] == "read_only" + + + +def test_policy_engine_requires_approval_for_clear_dtc(): + decision = policy.evaluate_request( + command_key="clear_trouble_codes", + payload={"command_key": "clear_trouble_codes"}, + ) + + assert decision is not None + assert decision.decision == DECISION_REQUIRE_APPROVAL + assert decision.to_dict()["policy_class"] == "destructive" + assert "operator_approval" in decision.to_dict()["guard_tags"] + + + +def test_policy_engine_blocks_force_and_raw_command_inputs(): + force_decision = policy.evaluate_request( + command_key="engine_rpm", + payload={"command_key": "engine_rpm", "force": True}, + ) + raw_decision = policy.evaluate_request( + command_key="engine_rpm", + payload={"command_key": "engine_rpm", "raw_command": "04"}, + ) + + assert force_decision is not None + assert force_decision.decision == DECISION_DENY + assert force_decision.to_dict()["action_key"] == "force_query" + + assert raw_decision is not None + assert raw_decision.decision == DECISION_DENY + assert raw_decision.to_dict()["action_key"] == "custom_command" diff --git a/tests/test_obd_ai_server.py b/tests/test_obd_ai_server.py index 3734312a..a00f855d 100644 --- a/tests/test_obd_ai_server.py +++ b/tests/test_obd_ai_server.py @@ -91,6 +91,17 @@ def test_function_server_rejects_unknown_tools(): assert payload["error"]["code"] == "unknown_tool" +def test_function_server_surfaces_policy_guardrail_errors_for_blocked_actions(): + app = _server_app() + + app.call_tool("connect_vehicle", {}) + payload = app.call_tool("read_sensor", {"command_key": "clear_trouble_codes"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "approval_required" + assert payload["error"]["details"]["policy_class"] == "destructive" + + def test_http_server_exposes_tools_and_call_endpoint(): manager = OBDAISessionManager(obd_factory=FakeFactory(FakeConnection())) surface = OBDAIReadOnlyToolSurface(session_manager=manager) diff --git a/tests/test_obd_ai_session.py b/tests/test_obd_ai_session.py index 37c2cdab..4d0c4a9f 100644 --- a/tests/test_obd_ai_session.py +++ b/tests/test_obd_ai_session.py @@ -76,12 +76,21 @@ def test_session_query_only_uses_approved_catalog_commands(): assert factory.connection.query_calls == [(obd.commands.RPM, False)] -def test_session_query_rejects_unapproved_command_keys(): +def test_session_query_rejects_unknown_command_keys(): factory = FakeFactory() manager = OBDAISessionManager(obd_factory=factory) session = manager.open_session() with pytest.raises(KeyError, match="Unapproved command key"): + session.query("definitely_not_approved") + + +def test_session_query_rejects_destructive_policy_actions_without_approval(): + factory = FakeFactory() + manager = OBDAISessionManager(obd_factory=factory) + session = manager.open_session() + + with pytest.raises(PermissionError, match="explicit approval"): session.query("clear_trouble_codes") @@ -111,6 +120,7 @@ def test_session_lists_serialized_supported_command_descriptors(): assert rpm["supported"] is True assert speed["supported"] is False + assert rpm["policy_class"] == "read_only" assert rpm["obd_command"]["name"] == "RPM" @@ -154,3 +164,12 @@ def test_session_can_query_serialized_response_payloads(): assert payload["command"]["name"] == "RPM" assert payload["value"]["type"] == "quantity" assert payload["value"]["magnitude"] == 900 + + +def test_session_rejects_force_true_bypass_requests(): + factory = FakeFactory() + manager = OBDAISessionManager(obd_factory=factory) + session = manager.open_session() + + with pytest.raises(PermissionError, match="force=True"): + session.query("engine_rpm", force=True) diff --git a/tests/test_obd_ai_tools.py b/tests/test_obd_ai_tools.py index 7dc92bd9..026091fd 100644 --- a/tests/test_obd_ai_tools.py +++ b/tests/test_obd_ai_tools.py @@ -2,6 +2,7 @@ from obd.OBDResponse import OBDResponse, Status, StatusTest from obd.utils import OBDStatus +from obd_ai.audit import InMemoryAuditSink from obd_ai.session import OBDAISessionManager from obd_ai.tools import OBDAIReadOnlyToolSurface @@ -69,9 +70,9 @@ def __call__(self, **kwargs): return self.connection -def _surface_for(connection): +def _surface_for(connection, audit_sink=None): manager = OBDAISessionManager(obd_factory=FakeFactory(connection)) - return OBDAIReadOnlyToolSurface(session_manager=manager) + return OBDAIReadOnlyToolSurface(session_manager=manager, audit_sink=audit_sink) def _null_response(command): @@ -120,14 +121,55 @@ def test_read_sensor_returns_structured_success_for_approved_supported_pid(): assert payload["data"]["response"]["command"]["name"] == "RPM" +def test_read_sensor_requires_approval_for_destructive_clear_dtc_action(): + connection = FakeConnection() + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "clear_trouble_codes"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "approval_required" + assert payload["error"]["details"]["policy_class"] == "destructive" + assert payload["error"]["details"]["decision"] == "require_approval" + assert connection.query_calls == [] + + +def test_read_sensor_blocks_force_true_policy_bypass(): + connection = FakeConnection() + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "engine_rpm", "force": True}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "blocked_action" + assert payload["error"]["details"]["action_key"] == "force_query" + assert connection.query_calls == [] + + +def test_read_sensor_blocks_raw_custom_command_fields(): + connection = FakeConnection() + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "engine_rpm", "raw_command": "04"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "blocked_action" + assert payload["error"]["details"]["action_key"] == "custom_command" + assert connection.query_calls == [] + + def test_read_sensor_blocks_unapproved_command_keys(): surface = _surface_for(FakeConnection()) surface.connect_vehicle() - payload = surface.read_sensor({"command_key": "clear_trouble_codes"}) + payload = surface.read_sensor({"command_key": "definitely_not_approved"}) assert payload["ok"] is False assert payload["error"]["code"] == "unsupported_pid" + assert payload["error"]["details"]["action_key"] == "unknown_command" def test_read_sensor_returns_disconnected_adapter_error_without_active_session(): @@ -172,6 +214,29 @@ def test_read_sensor_returns_timeout_error_when_query_times_out(): assert payload["error"]["code"] == "timeout" +def test_tool_surface_emits_audit_events_for_policy_decisions_and_tool_results(): + audit_sink = InMemoryAuditSink() + surface = _surface_for(FakeConnection(), audit_sink=audit_sink) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "clear_trouble_codes"}) + + assert payload["ok"] is False + assert len(audit_sink.events) >= 3 + + policy_event = next(event for event in audit_sink.events if event["event_type"] == "policy_decision") + tool_event = audit_sink.events[-1] + + assert policy_event["requested_action"] == "clear_trouble_codes" + assert policy_event["decision"] == "require_approval" + assert policy_event["result"] == "blocked" + assert policy_event["policy"]["policy_class"] == "destructive" + + assert tool_event["event_type"] == "tool_invocation" + assert tool_event["tool"] == "read_sensor" + assert tool_event["error_code"] == "approval_required" + + def test_list_supported_commands_only_reports_approved_allowlist_keys(): connection = FakeConnection(support_overrides={"RPM": True, "SPEED": False}) surface = _surface_for(connection) @@ -181,8 +246,10 @@ def test_list_supported_commands_only_reports_approved_allowlist_keys(): assert payload["ok"] is True keys = {item["key"] for item in payload["data"]["commands"]} + rpm = next(item for item in payload["data"]["commands"] if item["key"] == "engine_rpm") assert "engine_rpm" in keys assert "vehicle_speed" in keys + assert rpm["policy_class"] == "read_only" assert "clear_trouble_codes" not in keys