DVP: 자율주행 차량이 AI Flight Recorder가 필요한 이유

Source: Dev.to

자율주행 차량이 치명적인 결정을 내릴 때, 우리는 무엇을 가지고 있는가?

DVP – Driving Vehicle Protocol Specification

대부분의 경우 답은 벤더의 사일로에 잠겨 있는 독점 데이터 — 존재한다면 말이다 입니다.
공공 도로에서 생사를 가르는 결정을 내리는 AI 시스템을 배치함에 따라, 항공 분야의 수세기 동안 이어온 책임 기준과 자동차 산업의 “우리를 믿어라” 접근 방식 사이의 격차는 점점 감당하기 어려워지고 있습니다.

이 글에서는 Driving Vehicle Protocol (DVP) 를 소개합니다 — 자율주행 차량의 AI 의사결정에 대한 변조 방지 감사 추적을 만들기 위한 개방형 표준입니다. 이를 항공기의 블랙박스 기록 방식을 자율주행 자동차 시대에 도입하는 것으로 생각하면 됩니다.

문제

• 현대 자율주행 차량은 시간당 테라바이트 단위의 데이터를 생성합니다 (LiDAR 포인트 클라우드, 카메라 피드, 레이더 반환 등).
• 핵심 격차: AI가 무엇을 결정했는지와 그 이유에 대한 표준화된, 변조 방지 기록이 없습니다.
• 각 제조사는 자체 로깅 형식(있는 경우)을 구현합니다.
• 사고가 발생하면 조사에 의존하는 요소:
  1. 공급업체 협조
  2. 독점 데이터 형식
  3. 로그가 변조되지 않았다는 신뢰

이는 항공 분야와 근본적으로 다릅니다. 항공에서는 비행 데이터 기록 장치가 엄격한 국제 표준(ED‑112A, TSO‑C124)을 따릅니다.

규제 압력

책임 환경

When a Level 3+ ADS is engaged, the manufacturer—not the driver—bears responsibility. Proving what the system did (or didn’t do) becomes legally critical.

DVP 맥락에서

DVP는 더 넓은 VAP (Verifiable AI Provenance) 프레임워크 내의 산업 애플리케이션 프로필이며, 알고리즘 트레이딩을 위한 VCP를 만든 동일한 프레임워크입니다.

핵심 통찰: 모든 것을 표준화할 필요는 없습니다 — 감사 인터페이스만 표준화하면 됩니다.

아키텍처 개요

┌─────────────────────────────────────────┐
│  Proprietary (Competition Zone)          │
│  ─────────────────────────────────────  │
│  • Sensor‑fusion algorithms              │
│  • Neural‑network architectures          │
│  • Path‑planning implementations          │
│  • Internal data formats                 │
└─────────────────────────────────────────┘
                    ↓ 
            Minimal Audit Interface
                    ↓
┌─────────────────────────────────────────┐
│  DVP Common Header (Audit Zone)          │
│  ─────────────────────────────────────  │
│  ① timestamp     – When                 │
│  ② event_type    – What category        │
│  ③ action        – What happened        │
│  ④ model_id      – Which AI model       │
│  ⑤ prev_hash     – Cryptographic chain  │
└─────────────────────────────────────────┘

이 구분이 중요한 이유

• 경쟁 지적 재산권 노출 없음
• 최소한의 통합 오버헤드
• 보편적인 감사 가능성

DVP 분류 체계 (Python 열거형)

from enum import Enum

class DVPEventType(Enum):
    # Perception Events
    PERCEPTION_OBJECT_DETECTED = "PRC_OBJ"
    PERCEPTION_LANE_DETECTED   = "PRC_LAN"
    PERCEPTION_SIGN_DETECTED   = "PRC_SGN"
    PERCEPTION_CONFIDENCE_LOW  = "PRC_LOW"

    # Planning Events
    PLANNING_PATH_GENERATED    = "PLN_PTH"
    PLANNING_MANEUVER_DECIDED  = "PLN_MNV"
    PLANNING_ROUTE_CHANGED     = "PLN_RTE"

    # Control Commands
    CONTROL_ACCELERATION       = "CTL_ACC"
    CONTROL_BRAKE              = "CTL_BRK"
    CONTROL_STEERING           = "CTL_STR"
    CONTROL_EMERGENCY          = "CTL_EMG"

    # System Events
    SYSTEM_MODE_CHANGE          = "SYS_MOD"
    SYSTEM_TAKEOVER_REQUEST     = "SYS_TOR"
    SYSTEM_FALLBACK_INITIATED   = "SYS_FLB"
    SYSTEM_ODD_BOUNDARY          = "SYS_ODD"   # Operational Design Domain

해시 체인을 이용한 변조 방지 (Python 로거)

import hashlib
import json
from nacl.signing import SigningKey
from datetime import datetime, timezone
import uuid

class DVPEventLogger:
    """
    DVP‑compliant event logger for autonomous driving systems.
    Maintains a cryptographic hash chain for tamper‑evidence.
    """

    GENESIS_HASH = "0" * 64

    def __init__(self, signing_key: SigningKey, vehicle_id: str, model_id: str):
        self.signing_key = signing_key
        self.vehicle_id = vehicle_id
        self.model_id = model_id
        self.prev_hash = self.GENESIS_HASH

    def _canonical_json(self, obj: dict) -> bytes:
        """RFC 8785‑compliant JSON canonicalization."""
        return json.dumps(
            obj,
            sort_keys=True,
            separators=(',', ':'),
            ensure_ascii=False
        ).encode('utf-8')

    def log_event(
        self,
        event_type: str,
        action: str,
        sensor_inputs: dict = None,
        decision_factors: dict = None,
        confidence: float = None
    ) -> dict:
        """
        Create a DVP‑compliant event record with hash‑chain linkage.

        Returns:
            Complete DVP event record with cryptographic signature.
        """
        # UUIDv7 ensures temporal ordering
        event_id = str(uuid.uuid7())
        timestamp = datetime.now(timezone.utc).isoformat()

        # Header (minimal required fields)
        header = {
            "EventID":   event_id,
            "Timestamp": timestamp,
            "VehicleID": self.vehicle_id,
            "EventType": event_type
        }

        # Payload (domain‑specific content)
        payload = {
            "Action":   action,
            "ModelID":  self.model_id
        }

        if sensor_inputs:
            payload["SensorInputs"] = sensor_inputs
        if decision_factors:
            payload["DecisionFactors"] = decision_factors
        if confidence is not None:
            payload["Confidence"] = confidence

        # Assemble full record
        record = {
            "Header":  header,
            "Payload": payload,
            "PrevHash": self.prev_hash
        }

        # Compute hash over canonical JSON
        record_bytes = self._canonical_json(record)
        record_hash = hashlib.sha256(record_bytes).hexdigest()

        # Sign the hash
        signature = self.signing_key.sign(record_hash.encode()).signature.hex()

        # Attach hash and signature
        record["Hash"] = record_hash
        record["Signature"] = signature

        # Update chain
        self.prev_hash = record_hash

        return record

요점

DVP는 최소한의, 공급업체에 구애받지 않는 감사 인터페이스를 제공한다:

1. 독점 IP를 보호하면서 얇고 표준화된 헤더만 노출한다.
2. 변조 방지를 보장하기 위해 암호학적 해시 체이닝 및 디지털 서명을 사용한다.
3. 산업 간 조사 촉진을 위해 규제기관, 보험사, 법원에 신뢰할 수 있고 비교 가능한 데이터 소스를 제공한다—항공 분야의 블랙박스와 유사하게.

DVP(또는 유사한 개방형 표준)를 채택하는 것은 자율주행 차량 배치에서 책임 격차를 해소하기 위한 구체적인 단계이다.

이벤트 해싱 및 보안 봉투 (Python)

# Calculate event hash (includes previous hash for chain)
hash_input = (
    self._canonical_json(header) +
    self._canonical_json(payload) +
    self.prev_hash.encode()
)
event_hash = hashlib.sha256(hash_input).hexdigest()

# Create security envelope
security = {
    "Version": "1.0",
    "PrevHash": self.prev_hash,
    "EventHash": event_hash,
    "HashAlgo": "SHA256",
    "SignAlgo": "ED25519"
}

# Sign the event hash
signature = self.signing_key.sign(event_hash.encode())
security["Signature"] = signature.signature.hex()

# Update chain state
self.prev_hash = event_hash

return {
    "Header": header,
    "Payload": payload,
    "Security": security
}

from nacl.signing import SigningKey

# Initialize logger for vehicle
signing_key = SigningKey.generate()
logger = DVPEventLogger(
    signing_key=signing_key,
    vehicle_id="VEH-2025-DEMO-001",
    model_id="perception_v3.2.1"
)

# Log perception event: pedestrian detected
event1 = logger.log_event(
    event_type="PRC_OBJ",
    action="PEDESTRIAN_DETECTED",
    sensor_inputs={
        "lidar_confidence": 0.94,
        "camera_confidence": 0.91,
        "radar_detected": True,
        "distance_m": 32.5,
        "velocity_ms": -1.2  # Approaching
    },
    decision_factors={
        "fusion_method": "late_fusion_v2",
        "classification": "pedestrian_adult"
    },
    confidence=0.93
)

# Log planning decision: emergency brake initiated
event2 = logger.log_event(
    event_type="CTL_EMG",
    action="EMERGENCY_BRAKE_INITIATED",
    sensor_inputs={
        "ttc_seconds": 2.1,          # Time to collision
        "current_speed_ms": 15.6
    },
    decision_factors={
        "trigger": "pedestrian_in_path",
        "brake_force_percent": 100,
        "predicted_stop_distance_m": 18.2
    },
    confidence=0.99
)

print(json.dumps(event2, indent=2))

샘플 출력

{
  "Header": {
    "EventID": "019416a8-7c3f-7000-8000-000000000001",
    "Timestamp": "2025-01-03T12:34:56.789012+00:00",
    "VehicleID": "VEH-2025-DEMO-001",
    "EventType": "CTL_EMG"
  },
  "Payload": {
    "Action": "EMERGENCY_BRAKE_INITIATED",
    "ModelID": "perception_v3.2.1",
    "SensorInputs": {
      "ttc_seconds": 2.1,
      "current_speed_ms": 15.6
    },
    "DecisionFactors": {
      "trigger": "pedestrian_in_path",
      "brake_force_percent": 100,
      "predicted_stop_distance_m": 18.2
    },
    "Confidence": 0.99
  },
  "Security": {
    "Version": "1.0",
    "PrevHash": "a3f2...previous event hash...",
    "EventHash": "7b4c...current event hash...",
    "HashAlgo": "SHA256",
    "SignAlgo": "ED25519",
    "Signature": "9d3e...64 byte signature..."
  }
}

왜 모든 이벤트를 직접 앵커링해야 할까?

자율주행차(AV)는 분당 수천 개의 이벤트를 생성합니다. 각 이벤트를 블록체인에 앵커링하면:

• 🚫 비용이 엄청나게 많이 듦
• 📶 대역폭을 많이 사용함
• ❌ 불필요함

대신 로컬 해시 체인을 사용하고 주기적으로 앵커링합니다.

로컬 처리 (연속)

Event₁ → Event₂ → Event₃ → … → Eventₙ
   ↓        ↓        ↓          ↓
[Hash Chain continuously validated locally]
   ↓        ↓        ↓          ↓
        Accumulated in local storage

주기적 앵커링 (예: 10분마다 또는 여행 종료 시)

         ┌─────────┐
         │  Root   │ ← 외부에 앵커링되는 것은 이 32‑byte 해시 하나뿐
         └────┬────┘
        ┌─────┴─────┐
      ┌─┴─┐       ┌─┴─┐
      │H₁₂│       │H₃₄│
      └─┬─┘       └─┬─┘
     ┌──┴──┐    ┌──┴──┐
    H(E₁) H(E₂) H(E₃) H(E₄)

Merkle Tree Builder (RFC 6962‑준수)

import hashlib
from typing import List

def build_merkle_tree(event_hashes: List[str]) -> str:
    """
    RFC 6962‑준수 Merkle 트리를 구성합니다.

    매개변수:
        event_hashes: 이벤트 해시 문자열 목록

    반환값:
        Merkle 루트 해시 (16진수 문자열)
    """
    if not event_hashes:
        raise ValueError("Cannot build tree from empty list")

    # Leaf nodes: RFC 6962에 따라 0x00 접두사
    leaves = [
        hashlib.sha256(b'\x00' + h.encode()).digest()
        for h in event_hashes
    ]

    # 다음 2의 거듭제곱까지 패딩 (마지막 leaf 복제)
    while len(leaves) & (len(leaves) - 1):
        leaves.append(leaves[-1])

    # 트리를 아래에서 위로 구축
    while len(leaves) > 1:
        next_level = []
        for i in range(0, len(leaves), 2):
            # Internal nodes: RFC 6962에 따라 0x01 접두사
            combined = b'\x01' + leaves[i] + leaves[i + 1]
            next_level.append(hashlib.sha256(combined).digest())
        leaves = next_level

    return leaves[0].hex()

예시: 이벤트 배치를 앵커링하기

batch_hashes = [
    event1["Security"]["EventHash"],
    event2["Security"]["EventHash"]
]
merkle_root = build_merkle_tree(batch_hashes)

print(f"Merkle Root: {merkle_root}")
# 이 32‑바이트 해시는 블록체인이나 TSA에 앵커될 수 있습니다

Requirement: DVP Implementation

• Automatic logging – Hash chain captures all decisions
• Event timestamping – UUIDv7 + ISO 8601 timestamps
• Operation duration – SYSTEM_MODE_CHANGE events
• Input data logging – SensorInputs field
• Reference database – ModelID + version tracking

DVP와 신흥 표준 정렬

UN GTR on ADS

• 시나리오 재구성 – 인식에서 행동까지의 전체 인과 사슬
• 고장‑모드 문서화 – SYSTEM_FALLBACK 이벤트와 그 이유
• 인간‑감독 증거 – TAKEOVER_REQUEST 및 응답 타이밍
• ODD 경계 준수 – Operational Design Domain 이벤트

SOTIF 준수

• 인식 신뢰도 수준
• 센서 불일치 이벤트
• 에지‑케이스 발생
• 시스템‑불확실성 인식

벤더‑중립 감사 레이어

• IP 보호 – 알고리즘은 독점적으로 유지됩니다
• 규정 준수 지원 – EU AI 법 시행에 대비
• 포렌식 지원 – 사고 재구성이 가능해짐
• 미래 대비 – 규제가 강화됨에 따라