ライブラリ

CBOR

概要

CBOR（Concise Binary Object Representation、RFC 8949）は、コンパクトなバイナリオブジェクト表現形式です。JSONとの互換性を保ちながら、より効率的なデータ表現を実現し、特にIoTデバイスやWebAuthnでの標準採用により注目されています。セキュリティアプリケーションやコンストレイントデバイスでの使用が継続的に増加しており、小さなコードサイズ、効率的なメッセージサイズ、バージョンネゴシエーション不要の拡張性を設計目標とする次世代のデータ交換形式として位置づけられています。

詳細

CBOR（RFC 8949、旧RFC 7049）は、極めて小さなコードサイズ、効率的なメッセージサイズ、バージョンネゴシエーション不要の拡張性を設計目標とするバイナリシリアライゼーション形式です。CoAP（Constrained Application Protocol）のデータシリアライゼーション層やCOSE（CBOR Object Signing and Encryption）メッセージの基盤として推奨され、WebAuthnのCTAP（Client-to-Authenticator Protocol）でも使用されています。メモリ、プロセッサー性能、命令セットが非常に限られたシステムをサポートし、制約デバイス向けに最適化された実装が多数提供されています。

主な特徴

• コンパクトなバイナリ形式: JSONより効率的なデータ表現
• 自己記述性: スキーマ不要でデータ構造を記述
• 拡張性: バージョンネゴシエーション不要で機能拡張
• 多言語対応: 豊富な実装が各プログラミング言語で利用可能
• 標準準拠: IETF RFC 8949による国際標準
• IoT最適化: 制約デバイス向けのゼロメモリフットプリント実装

メリット・デメリット

メリット

• JSONとの概念的互換性を保ちながらバイナリの効率性を実現
• IoTデバイスやWebAuthnでの標準採用による信頼性
• 制約デバイス向けのメモリ効率的な実装
• 自己記述性によりスキーマ定義不要
• 豊富なデータ型サポート（バイナリデータ、日付、タグ付きデータ等）
• 多言語実装による高いポータビリティ

デメリット

• JSONと比較して可読性が低い（バイナリ形式）
• JSONほど広く普及していないため、ツールサポートが限定的
• 実装の複雑さがJSONより高い
• デバッグ時にバイナリデータの直接確認が困難
• WebやAPIでの標準的な採用はJSONに劣る
• エコシステムやライブラリサポートがJSONより少ない

参考ページ

• CBOR 公式サイト
• RFC 8949: Concise Binary Object Representation (CBOR)
• CBOR 実装一覧

書き方の例

基本的なエンコード・デコード（Python）

import cbor2
import json

# 基本データのエンコード
data = {
    "name": "CBOR Example",
    "version": 1.0,
    "features": ["compact", "self-describing", "extensible"],
    "metadata": {
        "created": "2025-01-01",
        "binary_data": b"Hello, CBOR!"
    }
}

# CBORエンコード
cbor_data = cbor2.dumps(data)
print(f"CBOR size: {len(cbor_data)} bytes")

# JSONエンコード（比較用）
json_data = json.dumps(data, default=str).encode('utf-8')
print(f"JSON size: {len(json_data)} bytes")
print(f"CBOR効率: {len(cbor_data) / len(json_data):.2%}")

# CBORデコード
decoded_data = cbor2.loads(cbor_data)
print("Decoded data:", decoded_data)

ファイル操作とストリーミング

import cbor2
from datetime import datetime, timezone

# 大きなデータセットの例
large_dataset = {
    "sensors": [
        {
            "id": f"sensor_{i:04d}",
            "type": "temperature" if i % 2 == 0 else "humidity",
            "readings": [
                {
                    "timestamp": datetime.now(timezone.utc),
                    "value": 20.0 + (i % 50),
                    "unit": "°C" if i % 2 == 0 else "%"
                }
                for _ in range(100)
            ]
        }
        for i in range(100)
    ],
    "metadata": {
        "collection_time": datetime.now(timezone.utc),
        "format_version": "1.2.0"
    }
}

# ファイルへの保存
with open('sensor_data.cbor', 'wb') as f:
    cbor2.dump(large_dataset, f)

# ファイルからの読み込み
with open('sensor_data.cbor', 'rb') as f:
    loaded_dataset = cbor2.load(f)

print(f"Loaded {len(loaded_dataset['sensors'])} sensors")

# ストリーミングエンコード（大きなデータ向け）
import io

def stream_encode_sensors(sensors):
    """センサーデータのストリーミングエンコード"""
    buffer = io.BytesIO()
    
    # ヘッダー情報をエンコード
    header = {"format": "sensor_stream", "version": "1.0"}
    buffer.write(cbor2.dumps(header))
    
    # 各センサーデータを個別にエンコード
    for sensor in sensors:
        sensor_cbor = cbor2.dumps(sensor)
        # 長さプレフィックス付きでエンコード
        buffer.write(cbor2.dumps(len(sensor_cbor)))
        buffer.write(sensor_cbor)
    
    return buffer.getvalue()

# ストリーミングデコード
def stream_decode_sensors(cbor_stream):
    """センサーデータのストリーミングデコード"""
    stream = io.BytesIO(cbor_stream)
    
    # ヘッダー読み込み
    header = cbor2.load(stream)
    print(f"Stream format: {header}")
    
    sensors = []
    while stream.tell() < len(cbor_stream):
        try:
            # データ長を読み込み
            data_length = cbor2.load(stream)
            # センサーデータを読み込み
            sensor_data = stream.read(data_length)
            sensor = cbor2.loads(sensor_data)
            sensors.append(sensor)
        except:
            break
    
    return sensors

IoTデバイス向け最適化（C風のPython実装）

import cbor2
import struct
from typing import Dict, Any

class IoTCBOREncoder:
    """IoTデバイス向けの最適化されたCBORエンコーダー"""
    
    @staticmethod
    def encode_sensor_reading(sensor_id: int, value: float, timestamp: int) -> bytes:
        """センサー読み値の効率的なエンコード"""
        # コンパクトなデータ構造
        compact_data = [
            sensor_id,      # 整数
            int(value * 100),  # 小数点2桁の固定小数点表現
            timestamp       # Unixタイムスタンプ
        ]
        
        return cbor2.dumps(compact_data)
    
    @staticmethod
    def encode_device_status(device_id: int, status: Dict[str, Any]) -> bytes:
        """デバイス状態の効率的なエンコード"""
        # 24キー以下の場合、単一バイトでキーをエンコード可能
        compact_status = {
            1: status.get('battery', 0),      # バッテリー残量(%)
            2: status.get('signal', 0),       # 信号強度
            3: status.get('temperature', 0),  # 内部温度
            4: int(status.get('online', False))  # オンライン状態
        }
        
        return cbor2.dumps([device_id, compact_status])

class IoTCBORDecoder:
    """IoTデバイス向けの最適化されたCBORデコーダー"""
    
    @staticmethod
    def decode_sensor_reading(cbor_data: bytes) -> Dict[str, Any]:
        """センサー読み値のデコード"""
        compact_data = cbor2.loads(cbor_data)
        
        return {
            'sensor_id': compact_data[0],
            'value': compact_data[1] / 100.0,
            'timestamp': compact_data[2]
        }
    
    @staticmethod
    def decode_device_status(cbor_data: bytes) -> Dict[str, Any]:
        """デバイス状態のデコード"""
        device_id, compact_status = cbor2.loads(cbor_data)
        
        return {
            'device_id': device_id,
            'battery': compact_status.get(1, 0),
            'signal': compact_status.get(2, 0),
            'temperature': compact_status.get(3, 0),
            'online': bool(compact_status.get(4, 0))
        }

# 使用例
encoder = IoTCBOREncoder()
decoder = IoTCBORDecoder()

# センサーデータのエンコード
sensor_cbor = encoder.encode_sensor_reading(
    sensor_id=101,
    value=23.45,
    timestamp=1640995200
)

print(f"Sensor data CBOR size: {len(sensor_cbor)} bytes")

# デコード
decoded_sensor = decoder.decode_sensor_reading(sensor_cbor)
print("Decoded sensor data:", decoded_sensor)

# デバイス状態のエンコード
status_cbor = encoder.encode_device_status(
    device_id=42,
    status={
        'battery': 85,
        'signal': -65,
        'temperature': 35.2,
        'online': True
    }
)

print(f"Status data CBOR size: {len(status_cbor)} bytes")
decoded_status = decoder.decode_device_status(status_cbor)
print("Decoded status:", decoded_status)

WebAuthn/CTAP統合

import cbor2
import hashlib
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec

class WebAuthnCBOR:
    """WebAuthn/CTAP向けのCBOR処理"""
    
    @staticmethod
    def create_webauthn_assertion(credential_id: bytes, 
                                signature: bytes, 
                                user_handle: bytes = None) -> bytes:
        """WebAuthn認証アサーションの作成"""
        assertion_data = {
            1: credential_id,  # credentialId
            2: signature,      # signature
        }
        
        if user_handle:
            assertion_data[3] = user_handle  # userHandle
        
        # CTAPの規定に従い、カノニカルCBORエンコーディングを使用
        return cbor2.dumps(assertion_data, canonical=True)
    
    @staticmethod
    def parse_webauthn_assertion(cbor_data: bytes) -> dict:
        """WebAuthn認証アサーションの解析"""
        data = cbor2.loads(cbor_data)
        
        return {
            'credential_id': data.get(1),
            'signature': data.get(2),
            'user_handle': data.get(3)
        }
    
    @staticmethod
    def create_ctap_request(command: int, parameters: dict) -> bytes:
        """CTAP リクエストの作成"""
        # CTAPコマンド構造
        ctap_data = {
            0x01: command,     # CTAP command
            0x02: parameters   # Parameters
        }
        
        return cbor2.dumps(ctap_data, canonical=True)

# WebAuthn使用例
webauthn = WebAuthnCBOR()

# 認証アサーションの作成
credential_id = b"example_credential_id_1234567890"
signature = b"example_signature_data_abcdef"
user_handle = b"user123"

assertion_cbor = webauthn.create_webauthn_assertion(
    credential_id=credential_id,
    signature=signature,
    user_handle=user_handle
)

print(f"WebAuthn assertion CBOR size: {len(assertion_cbor)} bytes")

# アサーションの解析
parsed_assertion = webauthn.parse_webauthn_assertion(assertion_cbor)
print("Parsed assertion:", parsed_assertion)

# CTAP リクエストの例
ctap_request = webauthn.create_ctap_request(
    command=0x01,  # authenticatorMakeCredential
    parameters={
        "clientDataHash": hashlib.sha256(b"client_data").digest(),
        "rp": {"id": "example.com", "name": "Example Corp"},
        "user": {"id": b"user123", "name": "[email protected]"},
        "pubKeyCredParams": [{"type": "public-key", "alg": -7}]
    }
)

print(f"CTAP request size: {len(ctap_request)} bytes")

高度なCBORタグとカスタム型

import cbor2
from datetime import datetime, timezone
import uuid
import base64

# カスタムエンコーダーの定義
def encode_datetime(encoder, value):
    """日時のカスタムエンコード（Tag 1: Epoch-based date/time）"""
    encoder.encode(cbor2.CBORTag(1, value.timestamp()))

def encode_uuid(encoder, value):
    """UUIDのカスタムエンコード（Tag 37: UUID）"""
    encoder.encode(cbor2.CBORTag(37, value.bytes))

def encode_regex(encoder, value):
    """正規表現のカスタムエンコード（Tag 35: Regular expression）"""
    encoder.encode(cbor2.CBORTag(35, value.pattern))

# カスタムデコーダーの定義
def decode_datetime(decoder, tag):
    """日時のカスタムデコード"""
    return datetime.fromtimestamp(tag.value, timezone.utc)

def decode_uuid(decoder, tag):
    """UUIDのカスタムデコード"""
    return uuid.UUID(bytes=tag.value)

def decode_regex(decoder, tag):
    """正規表現のカスタムデコード"""
    import re
    return re.compile(tag.value)

# エンコーダー・デコーダーの設定
encoder = cbor2.CBOREncoder()
encoder.encode_type(datetime, encode_datetime)
encoder.encode_type(uuid.UUID, encode_uuid)

decoder = cbor2.CBORDecoder()
decoder.tag_hook = {
    1: decode_datetime,
    37: decode_uuid,
    35: decode_regex
}

# 使用例
complex_data = {
    "id": uuid.uuid4(),
    "created_at": datetime.now(timezone.utc),
    "expires_at": datetime(2025, 12, 31, 23, 59, 59, tzinfo=timezone.utc),
    "data": {
        "numbers": [1, 2, 3.14159, -42],
        "binary": b"binary_data_example",
        "nested": {
            "flag": True,
            "null_value": None
        }
    }
}

# カスタム型を含むエンコード
import io
buffer = io.BytesIO()
encoder_instance = cbor2.CBOREncoder(buffer)
encoder_instance.encode_type(datetime, encode_datetime)
encoder_instance.encode_type(uuid.UUID, encode_uuid)
encoder_instance.encode(complex_data)

cbor_data = buffer.getvalue()
print(f"Complex data CBOR size: {len(cbor_data)} bytes")

# カスタム型を含むデコード
buffer = io.BytesIO(cbor_data)
decoder_instance = cbor2.CBORDecoder(buffer)
decoder_instance.tag_hook = {
    1: decode_datetime,
    37: decode_uuid
}

decoded_data = decoder_instance.decode()
print("Decoded complex data:", decoded_data)
print(f"UUID type: {type(decoded_data['id'])}")
print(f"DateTime type: {type(decoded_data['created_at'])}")

パフォーマンス比較とベンチマーク

import cbor2
import json
import pickle
import time
import sys
from typing import Any, Dict

class SerializationBenchmark:
    """シリアライゼーション形式のベンチマーク"""
    
    @staticmethod
    def create_test_data(size: int = 1000) -> Dict[str, Any]:
        """テストデータの生成"""
        return {
            "metadata": {
                "version": "1.0.0",
                "created": datetime.now().isoformat(),
                "description": "Benchmark test data"
            },
            "data": [
                {
                    "id": i,
                    "name": f"item_{i:05d}",
                    "value": i * 3.14159,
                    "active": i % 2 == 0,
                    "tags": [f"tag_{j}" for j in range(i % 5 + 1)],
                    "metadata": {
                        "category": "test",
                        "priority": i % 10,
                        "binary_data": f"data_{i}".encode('utf-8')
                    }
                }
                for i in range(size)
            ]
        }
    
    @staticmethod
    def benchmark_serialization(data: Any, iterations: int = 100):
        """シリアライゼーション形式のベンチマーク"""
        results = {}
        
        # CBOR
        start_time = time.time()
        for _ in range(iterations):
            cbor_data = cbor2.dumps(data)
        cbor_time = time.time() - start_time
        cbor_size = len(cbor_data)
        
        # JSON
        start_time = time.time()
        for _ in range(iterations):
            json_data = json.dumps(data, default=str).encode('utf-8')
        json_time = time.time() - start_time
        json_size = len(json_data)
        
        # Pickle
        start_time = time.time()
        for _ in range(iterations):
            pickle_data = pickle.dumps(data)
        pickle_time = time.time() - start_time
        pickle_size = len(pickle_data)
        
        results = {
            'CBOR': {'time': cbor_time, 'size': cbor_size, 'data': cbor_data},
            'JSON': {'time': json_time, 'size': json_size, 'data': json_data},
            'Pickle': {'time': pickle_time, 'size': pickle_size, 'data': pickle_data}
        }
        
        return results

# ベンチマーク実行
benchmark = SerializationBenchmark()
test_data = benchmark.create_test_data(1000)

print("Serialization Benchmark Results:")
print("=" * 50)

results = benchmark.benchmark_serialization(test_data, iterations=100)

for format_name, metrics in results.items():
    print(f"{format_name}:")
    print(f"  Time: {metrics['time']:.4f} seconds")
    print(f"  Size: {metrics['size']:,} bytes")
    print(f"  Efficiency: {metrics['size'] / results['JSON']['size']:.2%} of JSON")
    print()

# デシリアライゼーションベンチマーク
print("Deserialization Benchmark:")
print("=" * 30)

for format_name, metrics in results.items():
    start_time = time.time()
    
    for _ in range(100):
        if format_name == 'CBOR':
            cbor2.loads(metrics['data'])
        elif format_name == 'JSON':
            json.loads(metrics['data'].decode('utf-8'))
        elif format_name == 'Pickle':
            pickle.loads(metrics['data'])
    
    deserialize_time = time.time() - start_time
    print(f"{format_name} deserialization: {deserialize_time:.4f} seconds")