LibOSDP: Open Supervised Device Protocol Implementation

repository·master·Indexed 41 days ago

https://github.com/osdp-dev/libosdp

A cross-platform, open-source implementation of the IEC 60839-11-5 Open Supervised Device Protocol (OSDP) version 4.0.0.dev0. It supports Secure Channel (AES-128) communication to improve interoperability between access control and security products. The library provides C and Python APIs for implementing both Control Panel (CP) and Peripheral Device (PD) roles, with specific integration guides for ESP-IDF and PlatformIO.

Tokens
4.6K
Snippets
14
Records
22
Agent score
52%

What's inside LibOSDP

  1. Memory management for commands and events in embedded environments

    master

    In LibOSDP, commands and events are queued by reference. This means any object you submit to the library must remain alive and unmodified until the completion callback is triggered and returns ownership to you.

    For resource-constrained environments without a heap, follow this pattern to avoid memory corruption:

    1. Use a single static command/event object.
    2. Use an in-flight flag to track the status.
    3. Do not refill or modify the object while the flag is set.
    4. Clear the in-flight flag only inside the completion callback.

    For platforms with an allocator, you may alternatively use an 'allocate-per-submission' pattern as seen in the examples/c directory.

  2. How LibOSDP works: Channels, Commands, and Events

    master

    LibOSDP implements the OSDP protocol by managing communication between a Control Panel (CP) and one or more Peripheral Devices (PD). Your application interacts with the library through three primary constructs:

    • Channel: The transport layer that enables communication between two OSDP devices.
    • Commands: Actions initiated by a CP to a PD.
    • Events: Actions or notifications initiated by a PD to a CP.

    To use the library, you must:

    1. Implement the osdp_channel interface.
    2. Describe the device using osdp_pd_info_t.
    3. Create an osdp_t context.
    4. Call osdp_cp/pd_refresh() at least once every 50ms to satisfy protocol timing requirements.
  3. Threading and Synchronization in LibOSDP

    master
    LibOSDP is not internally synchronized. To prevent race conditions and ensure thread safety, you must serialize every call made against a single context onto a single task. Refer to the osdp_pd_refresh() documentation in osdp.h for specific lifecycle details.
  4. Memory ownership and callback patterns in LibOSDP

    master

    LibOSDP uses a specific ownership model for commands and events: they are queued by reference. To prevent memory leaks or use-after-free errors, follow these rules:

    1. Lifetime: A submitted object (command or event) must outlive the initial function call.
    2. Handover: The object is handed back to the application exactly once via its completion callback.
    3. Allocation Pattern: The safest pattern for production is to allocate the object per submission and free it inside the completion callback. This ensures the library no longer holds the object before it is destroyed.
  5. How the libosdp Python API manages concurrency

    master

    The Python library is a high-level wrapper around the C library. It abstracts away several low-level requirements of the C API:

    • Memory Management: Unlike the C library, which requires callers to keep objects alive manually, the Python library manages object lifecycles automatically.
    • Thread Safety and Serialization: The C library requires calls against a device to be serialized against its refresh loop. The Python library handles this in the background by owning the refresh thread and serializing API calls against it. This makes every call safe to make from any thread.
    • Non-blocking Status Checks: Status checks like is_online() and is_sc_active() are implemented to be lock-free, meaning polling these values will not wait on the serial bus.

    As a result, commands and events are treated as ordinary Python objects that you can build, submit, and forget.

  6. Implement a custom Channel for LibOSDP

    master

    LibOSDP does not manage the physical transport layer. You must implement a subclass of osdp.Channel to bridge the library to your hardware (e.g., Serial, TCP/IP). Your implementation must provide read(max_bytes), write(data), and flush() methods.

    import serial
    from osdp import Channel
    
    class SerialChannel(Channel):
        def __init__(self, device: str, speed: int = 115200):
            self.dev = serial.Serial(device, speed, timeout=0)
    
        def read(self, max_bytes: int) -> bytes:
            return self.dev.read(max_bytes)
    
        def write(self, data: bytes) -> int:
            return self.dev.write(data)
    
        def flush(self) -> None:
            self.dev.flush()
    
        def __del__(self):
            self.dev.close()
  7. Use Control Panel (CP) Mode

    master

    In Control Panel mode, the application acts as the central controller managing one or more Peripheral Devices (PDs).

    Key workflow:

    1. Define PDInfo objects for each device, including their address and a Channel.
    2. Initialize ControlPanel with the list of PDInfo objects.
    3. Call .start() to begin the handler thread.
    4. Call .sc_wait_all() to block until secure channels are established.
    5. Use .get_event(address) to retrieve events from specific PDs.
    6. Use .submit_command(address, command) to send commands to PDs.

    Commands and events are implemented as typed dataclasses, allowing for pattern matching (e.g., match cp.get_event(...)).

    from osdp import ControlPanel, PDInfo, KeyStore, LogLevel, LEDColor, commands, events
    
    # Setup channel and PD info
    channel = SerialChannel("/dev/ttyUSB0")
    pd_info = [PDInfo(101, channel, scbk=KeyStore.gen_key())]
    
    # Start CP
    cp = ControlPanel(pd_info, log_level=LogLevel.Debug)
    cp.start()
    cp.sc_wait_all()
    
    # Example: Send LED command
    led_cmd = commands.LED(
        reader=1,
        led_number=0,
        permanent=commands.PermanentLEDParams(
            on_color=LEDColor.Red,
            off_color=LEDColor.Black,
            on_count=10,
            off_count=10,
        ),
    )
    cp.submit_command(pd_info[0].address, led_cmd)
    
    # Example: Handle events
    while True:
        match cp.get_event(pd_info[0].address):
            case events.CardRead(data=data):
                print(f"CP: card {data.hex()}")
            case events.KeyPress(data=keys):
                print(f"CP: keypad {keys!r}")
  8. Use Peripheral Device (PD) Mode

    master

    In Peripheral Device mode, the application acts as an OSDP endpoint.

    Key workflow:

    1. Define PDInfo with a Channel and an optional scbk (Secure Channel Base Key). If scbk is None, secure channel is disabled.
    2. Define PDCapabilities to describe what the device can do.
    3. Implement a command_handler(cmd: Command) -> Command | None function. This function should:
      • Return a Command to answer a request inline.
      • Return None to accept a command without an immediate response.
      • Raise NakError to decline a command.
    4. Initialize PeripheralDevice with pd_info, pd_cap, and your command_handler.
    5. Call .start() and .sc_wait() to begin operation.
    6. Use .submit_event(event) to send events (like CardRead) to the CP.
    from osdp import (
        Capability, CardFormat, Command, NakCode, NakError, PDCapabilities, PDInfo,
        PeripheralDevice, commands, events
    )
    
    channel = SerialChannel("/dev/ttyUSB0")
    pd_info = PDInfo(101, channel, scbk=None)
    
    pd_cap = PDCapabilities([
        (Capability.OutputControl, 1, 1),
        (Capability.LEDControl, 2, 1),
    ])
    
    def command_handler(cmd: Command) -> Command | None:
        match cmd:
            case commands.Status(type=report_type):
                return commands.Status(type=report_type, report=read_inputs())
            case commands.BioRead():
                raise NakError(NakCode.BioType)
        return None
    
    pd = PeripheralDevice(pd_info, pd_cap, command_handler=command_handler)
    pd.start()
    pd.sc_wait()
    
    # Sending an event
    card_event = events.CardRead(
        reader_no=1,
        direction=1,
        format=CardFormat.ASCII,
        data=bytes([9, 1, 9, 2, 6, 3, 1, 7, 7, 0]),
    )
    pd.submit_event(card_event)
  9. Configure entropy for Secure Channel with TinyAES

    master

    When using the TinyAES backend for Secure Channel, LibOSDP uses rand_u32(), which is backed by esp_random(). On ESP32, esp_random() provides true hardware entropy only when Wi-Fi, Bluetooth, or the SAR ADC entropy source is active.

    If your application runs Secure Channel over TinyAES without using RF (Wi-Fi/BT), you must manually enable the entropy source to avoid using the predictable libc rand() fallback.

    Important: The entropy source cannot be shared. You must call bootloader_random_enable() before osdp_pd_setup() and call bootloader_random_disable() before starting Wi-Fi or Bluetooth later.

    // Example sequence for non-RF applications using TinyAES
    bootloader_random_enable();
    osdp_pd_setup(...);
    
    // ... later, before starting Wi-Fi/BT
    bootloader_random_disable();
  10. Test CP and PD interaction using socat pseudo-terminals

    master

    To simulate communication between a CP and a PD on a single machine without physical hardware, use socat to create a pair of linked pseudo-terminal devices. This creates a virtual serial channel.

    1. Create the linked pseudo-terminals:
    socat pty,raw,echo=0,nonblock,link=/tmp/ttyS0 pty,raw,echo=0,nonblock,link=/tmp/ttyS1
    1. In a separate terminal, start the CP application using the first device:
    ./examples/python/cp_app.py /tmp/ttyS0
    1. Start the PD application using the second device:
    ./examples/python/pd_app.py /tmp/ttyS1
    socat pty,raw,echo=0,nonblock,link=/tmp/ttyS0 pty,raw,echo=0,nonblock,link=/tmp/ttyS1
    
    ./examples/python/cp_app.py /tmp/ttyS0
    ./examples/python/pd_app.py /tmp/ttyS1