From 576f22268f996c4a0cbb07f585cb0f17978d47fb Mon Sep 17 00:00:00 2001 From: scheianu Date: Thu, 1 Oct 2026 15:31:04 +0300 Subject: [PATCH] RDM card reader: print card events, select the right port, serve taps to a web app Card reading (rdm6300 + pyserial) printing card inserted/removed with the full card data, plus a --serve mode that publishes taps over HTTP so a web page can authenticate a tap. The bridge reports card ids only - mapping a card to a user stays on the server. The tap is latched for --tap-ttl seconds, including after the card is lifted, and is single use, so a page reload cannot replay one tap into two logins. Port selection identifies the reader by serial number or by-id path instead of /dev/ttyUSBn, and refuses to guess when a selector matches several adapters. deploy/ has the systemd unit, a udev rule for a stable /dev/rdm-reader, and install.sh. Covered by unit tests plus PTY end-to-end tests. --- .gitignore | 12 + LICENSE | 21 ++ README.md | 376 ++++++++++++++++++++++++ deploy/99-rdm-reader.rules | 28 ++ deploy/install.sh | 162 ++++++++++ deploy/rdm-card-reader.service | 53 ++++ pyproject.toml | 40 +++ src/rdm_card_reader/__init__.py | 51 ++++ src/rdm_card_reader/__main__.py | 8 + src/rdm_card_reader/bridge.py | 503 ++++++++++++++++++++++++++++++++ src/rdm_card_reader/card.py | 135 +++++++++ src/rdm_card_reader/cli.py | 285 ++++++++++++++++++ src/rdm_card_reader/demo.py | 49 ++++ src/rdm_card_reader/ports.py | 397 +++++++++++++++++++++++++ src/rdm_card_reader/py.typed | 0 src/rdm_card_reader/reader.py | 111 +++++++ tests/__init__.py | 0 tests/conftest.py | 8 + tests/test_bridge.py | 501 +++++++++++++++++++++++++++++++ tests/test_card.py | 111 +++++++ tests/test_cli.py | 93 ++++++ tests/test_e2e_bridge.py | 177 +++++++++++ tests/test_e2e_pty.py | 165 +++++++++++ tests/test_ports.py | 228 +++++++++++++++ tests/test_protocol.py | 108 +++++++ tests/test_reader.py | 149 ++++++++++ 26 files changed, 3771 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 deploy/99-rdm-reader.rules create mode 100755 deploy/install.sh create mode 100644 deploy/rdm-card-reader.service create mode 100644 pyproject.toml create mode 100644 src/rdm_card_reader/__init__.py create mode 100644 src/rdm_card_reader/__main__.py create mode 100644 src/rdm_card_reader/bridge.py create mode 100644 src/rdm_card_reader/card.py create mode 100644 src/rdm_card_reader/cli.py create mode 100644 src/rdm_card_reader/demo.py create mode 100644 src/rdm_card_reader/ports.py create mode 100644 src/rdm_card_reader/py.typed create mode 100644 src/rdm_card_reader/reader.py create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/test_bridge.py create mode 100644 tests/test_card.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_e2e_bridge.py create mode 100644 tests/test_e2e_pty.py create mode 100644 tests/test_ports.py create mode 100644 tests/test_protocol.py create mode 100644 tests/test_reader.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..218f8d2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.venv/ +venv/ +.pytest_cache/ +.coverage +htmlcov/ +.idea/ +.vscode/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..14fac91 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..7e30a7a --- /dev/null +++ b/README.md @@ -0,0 +1,376 @@ +# rdm-card-reader + +Read 125 kHz RFID cards (RDM6300 reader, EM4100 cards) on a serial port and +print what happens: **card inserted** with the card data, then **card removed**. + +Built for a Raspberry Pi workstation, but it runs anywhere with a serial port. + +--- + +## What it does + +``` +$ rdm-card-reader + +-> Opening /dev/ttyAMA0 ... + OK reader ready on /dev/ttyAMA0 - waiting for cards (Ctrl+C to stop) + +---------------------------------------------------------- + CARD INSERTED - 2026-10-01 14:56:54 + id : 12886197 (0x00C4A0B5) + type : 0x0F + checksum : 222 + valid : True + port : /dev/ttyAMA0 +---------------------------------------------------------- + CARD REMOVED - 2026-10-01 14:56:54 id: 12886197 (0x00C4A0B5) +``` + +`id` is the number printed on the card. A card is reported once on insert +(the reader library de-duplicates repeated reads while the card sits on the +antenna) and once on removal. + +--- + +## Install + +```bash +pip install . +``` + +On a Raspberry Pi OS install with an externally-managed Python, add +`--break-system-packages` or use a virtual environment: + +```bash +python3 -m venv .venv && . .venv/bin/activate && pip install . +``` + +`rdm6300` and `pyserial` are pulled in automatically. + +--- + +## Usage + +```bash +rdm-card-reader # auto-detect the serial port +rdm-card-reader --port /dev/ttyUSB0 # use a specific port +rdm-card-reader --json # one JSON object per event +rdm-card-reader --heartbeat 1.5 # slower card-removed detection +rdm-card-reader --list-ports # show which ports exist, then exit +rdm-card-reader --serve # serve card data to a web app (see below) +rdm-card-reader --demo # simulate reads, no hardware needed +``` + +Also works as a module: `python -m rdm_card_reader`. + +| Option | Default | Meaning | +| --- | --- | --- | +| `--port` | auto-detect | Serial port: device path, by-id symlink, serial number or `vid:pid` | +| `--autodetect` | off | Pick the only USB adapter; refuse if there is more than one | +| `--heartbeat` | `0.5` | Seconds without a read before the card counts as removed | +| `--json` | off | Machine-readable output instead of the text block | +| `--list-ports` | – | List candidate and identified ports, then exit | +| `--verbose-ports` | off | With `--list-ports`, show by-id / by-path detail | +| `--demo` | – | Simulated reads; needs no hardware and no `rdm6300` | +| `--demo-count` | `3` | Number of simulated reads in `--demo` | +| `--version` | – | Print the version | + +With `--serve`: `--host` (`127.0.0.1`), `--bridge-port` (`8765`), `--mode` +(`login`), `--tap-ttl` (`10`), `--allow-origin` (repeatable), `--token`, +`--allow-remote`, `--verbose`. + +When `--port` is omitted the ports are tried in this order, the same order the +workstation app uses: + +``` +/dev/ttyS0 -> /dev/ttyAMA0 -> /dev/ttyUSB0 -> /dev/ttyACM0 +``` + +### Output modes + +Text mode is for humans and logs. `--json` prints one object per event, which +is what you want when piping into another program: + +```json +{"event": "card_inserted", "timestamp": "2026-10-01 14:56:54", "id": 12886197, "hex": "0x00C4A0B5", "type": 15, "checksum": 222, "is_valid": true, "port": "/dev/ttyAMA0"} +{"event": "card_removed", "timestamp": "2026-10-01 14:56:54", "id": 12886197, "hex": "0x00C4A0B5", "type": 15, "checksum": 222, "is_valid": true, "port": "/dev/ttyAMA0"} +``` + +### Exit codes + +| Code | Meaning | +| --- | --- | +| `0` | Ran normally, or stopped with Ctrl+C | +| `1` | No serial port could be opened | +| `2` | The `rdm6300` library is not installed | + +--- + +## Serving card data to a web app (`--serve`) + +This is the mode for **authenticating a web page**: the reader runs as a service +on the device, and the page asks it for the card that was just tapped. + +```bash +rdm-card-reader --serve --port /dev/rdm-reader --allow-origin https://your.web.app +``` + +The bridge reports **card ids only**. Deciding which user a card belongs to is +your server's job, not this program's. + +### The flow + +```mermaid +sequenceDiagram + participant C as Card + participant B as rdm-card-reader (Pi) + participant K as Web page + participant W as Your server + + C->>B: tap + B->>B: latch the tap (tap-ttl seconds) + K->>B: GET /api/card + B-->>K: {claimable: true, card: {id: ...}} + K->>B: POST /api/card/claim + B-->>K: {card: {id: ...}} (once) + K->>W: POST /login {card_id} + W->>W: card_id -> user + W-->>K: session +``` + +The point of the *latch* is that the tap stays available for `--tap-ttl` +seconds (default 10), **including after the card is lifted**. A quick tap +therefore cannot be lost because the page happened to poll a moment later. + +The point of *claim* is single use. Once a tap is spent, a page reload or a +second tab gets `409 Conflict` instead of a second login, so one physical tap +can never authenticate two sessions. + +### Endpoints + +| Method | Path | Purpose | +| --- | --- | --- | +| `GET` | `/api/card` | The latched tap: `present`, `claimable`, `consumed`, `age_seconds`, `card` | +| `POST` | `/api/card/claim` | Spend the tap on one login. `200`, or `409` if used/none, `410` if expired | +| `GET` | `/api/events` | Server-Sent Events stream, pushes each event as it happens | +| `GET` | `/api/last` | Most recent card event, or `204` | +| `GET` | `/api/taps` | Recent tap history (`?limit=N`) | +| `GET` | `/api/health` | Reader state, port, uptime, tap counts — for monitoring | +| `POST` | `/api/mode` | `{"mode": "login"}` or `{"mode": "monitor"}` | + +`login` mode (default) publishes taps only — a tap is an authentication +attempt. `monitor` mode also publishes removals, which is what you want for +watching a reader rather than logging people in. + +### Browser example + +```js +const state = await (await fetch("http://127.0.0.1:8765/api/card")).json(); +if (state.claimable) { + const res = await fetch("http://127.0.0.1:8765/api/card/claim", { method: "POST" }); + if (res.ok) { + const { card } = await res.json(); + await fetch("/login/card", { // your server decides who card.id is + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ card_id: card.id, hex: card.hex }), + }); + } +} +``` + +Or push instead of poll, so a tap logs in immediately: + +```js +const events = new EventSource("http://127.0.0.1:8765/api/events"); +events.onmessage = (e) => { const tap = JSON.parse(e.data); /* tap.id */ }; +``` + +### Security, and its limits + +Read :ref:`the warning below `: an EM4100 id is a +*username*, not a password. These controls bound who can *read* the bridge; +they do not make the card itself convincing as a sole credential. + +| Control | Behaviour | +| --- | --- | +| Bind address | `127.0.0.1` by default. Anything else needs `--allow-remote`, so an open card-event endpoint cannot appear on the LAN by accident. | +| Origins | Cross-origin browser requests are **refused** unless `--allow-origin` lists that exact origin. Stops another site from reading your taps. | +| Token | `--token` (or `RDM_BRIDGE_TOKEN`) is required from non-loopback callers. | +| Single use | A tap can only ever produce one successful claim. | +| Expiry | A tap older than `--tap-ttl` is refused with `410`. | + +Add a second factor — a PIN after the tap — wherever a cloned card would +actually cost you something. + + +> **Card ids are not secrets.** EM4100 cards broadcast a fixed 40-bit id with no +> challenge-response, and clones are cheap. If holding an id is enough to log +> in, the security is "whoever knows the number". Treat card auth as +> identification, and require something else for anything sensitive. + +--- + +## Running at boot (Raspberry Pi) + +```bash +sudo ./deploy/install.sh --origin https://your.web.app +``` + +That installs a service account, a virtualenv, a systemd unit and a udev rule, +then enables the service. Afterwards: + +```bash +systemctl status rdm-card-reader # state +journalctl -u rdm-card-reader -f # live log +curl http://127.0.0.1:8765/api/card # tap a card, then run this +``` + +### Identifying the right ttyUSB + +`/dev/ttyUSB0` is **not** a stable identity — unplug something and the number +changes. Point this at the wrong adapter and you authenticate the wrong reader, +so the tooling is deliberately strict about it. + +```bash +rdm-card-reader --list-ports --verbose-ports +``` + +``` +Identified serial ports (best candidate first): + [score 100] /dev/ttyUSB0 :: CH340 :: 1a86:7523 :: S/N AB0XYZ12 + by-id : /dev/serial/by-id/usb-1a86_USB_Serial-if00-port0 + by-path : /dev/serial/by-path/pci-0000:01:00.0-usb-0:1.2:1.0-port0 +``` + +Then pin it, most specific first: + +```bash +--port /dev/rdm-reader # udev alias +--port /dev/serial/by-id/usb-1a86_USB_Serial-if00-port0 # by device +--port AB0XYZ12 # by serial number +--port 1a86:7523 # by chip +``` + +`--autodetect` picks the only USB adapter, and **refuses** when there is more +than one — silently choosing between two readers is exactly the failure this +avoids. Verify the wiring in `deploy/99-rdm-reader.rules`. + +Exit codes: `0` ok, `1` no port, `2` library missing, `3` port ambiguous. + +--- + +## Wiring + +The RDM6300 talks plain serial at **9600 baud, 8N1**. Connect it to the Pi's +UART (pins 8/10 for TXD/RXD, 5 V and GND) or through a USB-to-serial adapter. + +On a Raspberry Pi you usually need to free up the primary UART first: + +```bash +sudo raspi-config # Interface Options -> Serial Port +# "login shell over serial?" -> No +# "serial port hardware?" -> Yes +sudo usermod -aG dialout $USER # then log out and back in +``` + +The GPIO UART appears as `/dev/serial0` (a symlink to `/dev/ttyAMA0` or +`/dev/ttyS0`). + +--- + +## The protocol + +Worth knowing if you ever decode the bytes yourself. The reader emits a frame +between `STX` (0x02) and `ETX` (0x03) as **12 ASCII hex nibbles**: + +``` +0x02 0F 00C4A0B5 DE 0x03 + | | | + | | +-- checksum + | +----------- card id (4 bytes, big-endian) + +-------------- type byte +``` + +Two details that are easy to get wrong: + +1. **Field order is type, id, checksum** — the id is *not* first. +2. **The checksum is the XOR of the type byte *and* all four id bytes.** XOR-ing + only the id bytes produces a mismatch on every card. + +``` +checksum = type ^ b0 ^ b1 ^ b2 ^ b3 + = 0x0F ^ 0x00 ^ 0xC4 ^ 0xA0 ^ 0xB5 + = 0xDE (= 222) +``` + +A frame whose checksum does not match is reported as `INVALID CARD` and is +never treated as a card insert. + +There is also a subtle failure mode: a frame with the wrong number of nibbles +is not rejected outright. Two stray nibbles stay in the buffer and corrupt the +*next* read instead of the current one, so a single glitch can look like a +problem with the following card. + +--- + +## Troubleshooting + +| Symptom | Fix | +| --- | --- | +| `device not found` on every port | Check the wiring; run `rdm-card-reader --list-ports` | +| `permission denied` | `sudo usermod -aG dialout $USER`, then log out and back in | +| Reader opens but no cards | Card must be EM4100 / 125 kHz; hold it ~2 cm from the coil | +| `INVALID CARD` repeatedly | Noisy supply or long/unshielded wiring; shorten it | +| Every card reported twice | Lower `--heartbeat` if the card is lifted quickly | + +--- + +## Development + +```bash +pip install -e ".[dev]" +pytest +``` + +The suite needs no RFID hardware: the end-to-end tests open a pseudo-terminal, +write real RDM6300 frames into it and assert on the CLI output. `--demo` also +works with no hardware at all. + +### Layout + +``` +src/rdm_card_reader/ + card.py data model, constants, output formatting (no hardware) + reader.py serial port handling, event callbacks + cli.py argument parsing and program flow + demo.py simulated reads +tests/ + test_protocol.py RDM6300 wire format, byte for byte + test_card.py event model and formatting + test_reader.py port selection, using a fake reader library + test_cli.py exit codes and output modes + test_e2e_pty.py full stack over a pseudo-terminal +``` + +### Using it as a library + +```python +from rdm_card_reader import open_reader + +def on_event(event): + if event.event == "card_inserted": + print("card", event.value, event.hex_value) + +reader = open_reader(on_event) +if reader is None: + raise SystemExit("no reader found") + +reader.start() # blocks; call reader.close() to stop +``` + +--- + +## License + +MIT diff --git a/deploy/99-rdm-reader.rules b/deploy/99-rdm-reader.rules new file mode 100644 index 0000000..909ee32 --- /dev/null +++ b/deploy/99-rdm-reader.rules @@ -0,0 +1,28 @@ +# Give the RFID reader a stable name: /dev/rdm-reader +# +# Without this, the adapter is /dev/ttyUSB0 or /dev/ttyUSB1 depending on the +# order the kernel enumerated USB devices, which changes between reboots and +# whenever anything else is plugged in. Authenticating against the wrong reader +# is a real failure, so pin the identity here instead of hoping. +# +# Match on the USB-serial bridge chip. Add serial numbers once you know them - +# see README "Identifying the right ttyUSB" for how to read them off: +# +# SUBSYSTEM=="tty", ATTRS{idVendor}=="1a86", ATTRS{idProduct}=="7523", \ +# ATTRS{serial}=="REPLACE_WITH_YOUR_SERIAL", SYMLINK+="rdm-reader", MODE="0660", GROUP="dialout" +# +# Current rule: match by vendor:product (CH340 is 1a86:7523). +# Commented-out alternatives for other common bridge chips are listed below. +# Only ONE rule should be active - remove or comment the others. + +# CH340 / CH341 (most common on cheap USB-TTL adapters) +SUBSYSTEM=="tty", ATTRS{idVendor}=="1a86", ATTRS{idProduct}=="7523", SYMLINK+="rdm-reader", MODE="0660", GROUP="dialout" + +# FTDI FT232R +# SUBSYSTEM=="tty", ATTRS{idVendor}=="0403", ATTRS{idProduct}=="6001", ATTRS{serial}=="REPLACE_ME", SYMLINK+="rdm-reader", MODE="0660", GROUP="dialout" + +# Silicon Labs CP2102 +# SUBSYSTEM=="tty", ATTRS{idVendor}=="10c4", ATTRS{idProduct}=="ea60", ATTRS{serial}=="REPLACE_ME", SYMLINK+="rdm-reader", MODE="0660", GROUP="dialout" + +# Prolific PL2303 +# SUBSYSTEM=="tty", ATTRS{idVendor}=="067b", ATTRS{idProduct}=="2303", ATTRS{serial}=="REPLACE_ME", SYMLINK+="rdm-reader", MODE="0660", GROUP="dialout" diff --git a/deploy/install.sh b/deploy/install.sh new file mode 100755 index 0000000..6c6134c --- /dev/null +++ b/deploy/install.sh @@ -0,0 +1,162 @@ +#!/usr/bin/env bash +# +# Install the RDM card reader as a service that starts at boot and keeps the +# card on the reader available over HTTP for the web page to claim. +# +# Run as root on the Raspberry Pi: +# +# sudo ./deploy/install.sh --origin https://your.web.app +# +# Options: +# --origin URL web app origin allowed to read the API (repeatable) +# --port PORT HTTP port for the bridge (default 8765) +# --dir PATH install location (default /opt/rdm-card-reader) +# --user NAME service account (default rdmcard) +# --no-service install and configure, but do not enable the service +# +set -euo pipefail + +ORIGINS=() +BRIDGE_PORT=8765 +INSTALL_DIR=/opt/rdm-card-reader +SERVICE_USER=rdmcard +INSTALL_SERVICE=1 +SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + +while [[ $# -gt 0 ]]; do + case "$1" in + --origin) ORIGINS+=("$2"); shift 2 ;; + --port) BRIDGE_PORT="$2"; shift 2 ;; + --dir) INSTALL_DIR="$2"; shift 2 ;; + --user) SERVICE_USER="$2"; shift 2 ;; + --no-service) INSTALL_SERVICE=0; shift ;; + -h|--help) sed -n '2,20p' "$0"; exit 0 ;; + *) echo "unknown option: $1" >&2; exit 2 ;; + esac +done + +log() { printf '\n\033[1m==> %s\033[0m\n' "$*"; } +warn() { printf '\033[33mwarning: %s\033[0m\n' "$*" >&2; } + +if [[ $EUID -ne 0 ]]; then + echo "Please run as root: sudo $0 $*" >&2 + exit 1 +fi + +# --------------------------------------------------------------------------- +# 1. System packages +# --------------------------------------------------------------------------- +log "Checking system packages" +apt-get update -qq +apt-get install -y --no-install-recommends python3 python3-venv python3-pip + +# --------------------------------------------------------------------------- +# 2. Install the application +# --------------------------------------------------------------------------- +log "Installing to $INSTALL_DIR" +mkdir -p "$INSTALL_DIR" + +# Copy the source tree (excluding local build/test artifacts). +for item in src tests pyproject.toml README.md LICENSE deploy; do + if [[ -e "$SOURCE_DIR/$item" ]]; then + cp -a "$SOURCE_DIR/$item" "$INSTALL_DIR/" + fi +done + +log "Building the virtual environment" +python3 -m venv "$INSTALL_DIR/.venv" +"$INSTALL_DIR/.venv/bin/pip" install --quiet --upgrade pip +"$INSTALL_DIR/.venv/bin/pip" install --quiet "$INSTALL_DIR" + +# --------------------------------------------------------------------------- +# 3. Service account +# --------------------------------------------------------------------------- +log "Creating service account '$SERVICE_USER'" +if ! id -u "$SERVICE_USER" >/dev/null 2>&1; then + # No login shell, no home directory: this account only reads a serial port. + useradd --system --no-create-home --shell /usr/sbin/nologin "$SERVICE_USER" +fi + +# Serial access requires the dialout group. +usermod -aG dialout "$SERVICE_USER" +echo " $SERVICE_USER is in group dialout" + +# --------------------------------------------------------------------------- +# 4. Stable device name +# --------------------------------------------------------------------------- +log "Installing the udev rule (/dev/rdm-reader)" +install -m 0644 "$SOURCE_DIR/deploy/99-rdm-reader.rules" \ + /etc/udev/rules.d/99-rdm-reader.rules +udevadm control --reload-rules +udevadm trigger --subsystem-match=tty || true +sleep 1 + +if [[ -e /dev/rdm-reader ]]; then + echo " /dev/rdm-reader -> $(readlink -f /dev/rdm-reader)" +else + warn "/dev/rdm-reader does not exist yet." + warn "The reader may not be plugged in, or it uses a different chip." + warn "Run '$INSTALL_DIR/.venv/bin/rdm-card-reader --list-ports --verbose-ports'" + warn "then edit /etc/udev/rules.d/99-rdm-reader.rules to match your adapter." +fi + +# --------------------------------------------------------------------------- +# 5. Service +# --------------------------------------------------------------------------- +log "Installing the systemd unit" +UNIT=/etc/systemd/system/rdm-card-reader.service +install -m 0644 "$SOURCE_DIR/deploy/rdm-card-reader.service" "$UNIT" + +# Rewrite the ExecStart line with the real install dir, port and origins. +ORIGIN_ARGS="" +for origin in "${ORIGINS[@]:-}"; do + [[ -n "$origin" ]] && ORIGIN_ARGS+=" --allow-origin $origin" +done + +python3 - "$UNIT" "$INSTALL_DIR" "$BRIDGE_PORT" "$ORIGIN_ARGS" "$SERVICE_USER" <<'PY' +import re, sys +unit, install_dir, port, origin_args, user = sys.argv[1:6] +text = open(unit).read() + +command = ( + f"{install_dir}/.venv/bin/rdm-card-reader --serve " + f"--port /dev/rdm-reader --bridge-port {port} --mode login{origin_args}" +) +# Replace the whole (possibly line-continued) ExecStart definition. +text = re.sub(r"ExecStart=[\s\S]*?(?=\n\n|\nRestart=)", f"ExecStart={command}", text, count=1) +text = re.sub(r"^User=.*$", f"User={user}", text, count=1, flags=re.M) +open(unit, "w").write(text) +PY + +# Make the venv readable by the service account. +chown -R "$SERVICE_USER:$SERVICE_USER" "$INSTALL_DIR" + +if [[ "$INSTALL_SERVICE" -eq 1 ]]; then + log "Enabling and starting the service" + systemctl daemon-reload + systemctl enable rdm-card-reader.service + systemctl restart rdm-card-reader.service + sleep 2 + systemctl --no-pager --lines=15 status rdm-card-reader.service || true +else + log "Skipping service start (--no-service)" +fi + +# --------------------------------------------------------------------------- +# Done +# --------------------------------------------------------------------------- +cat <=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "rdm-card-reader" +version = "0.2.0" +description = "RDM6300 / EM4100 RFID card reader: prints card events, or serves them over HTTP so a web app can authenticate a card tap." +readme = "README.md" +requires-python = ">=3.8" +license = { text = "MIT" } +keywords = ["rfid", "rdm6300", "em4100", "raspberry-pi", "serial"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Manufacturing", + "License :: OSI Approved :: MIT License", + "Operating System :: POSIX :: Linux", + "Programming Language :: Python :: 3", + "Topic :: System :: Hardware", +] +dependencies = [ + "rdm6300>=0.1.1", + "pyserial>=3.5", +] + +[project.optional-dependencies] +dev = ["pytest>=7.0"] + +[project.scripts] +rdm-card-reader = "rdm_card_reader.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +rdm_card_reader = ["py.typed"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" diff --git a/src/rdm_card_reader/__init__.py b/src/rdm_card_reader/__init__.py new file mode 100644 index 0000000..a02ce8f --- /dev/null +++ b/src/rdm_card_reader/__init__.py @@ -0,0 +1,51 @@ +"""Reading RDM6300 (EM4100) RFID cards as simple insert / remove events. + +Two ways to use this package: + +* as a plain reader that prints what happens (``rdm-card-reader``) +* as a small HTTP bridge that publishes card events for a web app + (``rdm-card-reader --serve``) + +The bridge reports card ids only. Deciding which user a card belongs to is the +server's job - see ``bridge`` for the reasoning. +""" + +from .card import ( + CANDIDATE_PORTS, + CONFIG_CARD_VALUE, + CardData, + CardEvent, + format_event, +) +from .bridge import EventHub, create_server +from .ports import ( + AmbiguousPort, + PortError, + PortInfo, + PortNotFound, + discover, + resolve, +) +from .reader import available, list_ports, open_reader + +__version__ = "0.2.0" + +__all__ = [ + "CANDIDATE_PORTS", + "CONFIG_CARD_VALUE", + "CardData", + "CardEvent", + "AmbiguousPort", + "EventHub", + "PortError", + "PortInfo", + "PortNotFound", + "available", + "create_server", + "discover", + "format_event", + "list_ports", + "open_reader", + "resolve", + "__version__", +] diff --git a/src/rdm_card_reader/__main__.py b/src/rdm_card_reader/__main__.py new file mode 100644 index 0000000..980febd --- /dev/null +++ b/src/rdm_card_reader/__main__.py @@ -0,0 +1,8 @@ +"""Allow ``python -m rdm_card_reader``.""" + +import sys + +from .cli import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/rdm_card_reader/bridge.py b/src/rdm_card_reader/bridge.py new file mode 100644 index 0000000..cfa01e9 --- /dev/null +++ b/src/rdm_card_reader/bridge.py @@ -0,0 +1,503 @@ +"""HTTP bridge: expose card taps to a web page, without ever becoming the +authority on who the card belongs to. + +The reader is a *sensor*. This module publishes what the sensor sees and stops +there - resolving a card id to a user is the server's job, so nothing here +writes to a database, mints session tokens or decides whether a login succeeds. + +Served by default: + +============================ ============================================ +``GET /api/health`` reader state, port, uptime, last tap age +``GET /api/card`` the card on the reader right now (latched) +``POST /api/card/claim`` claim that tap, single use, for one login +``GET /api/events`` Server-Sent Events stream of card events +``GET /api/last`` most recent card event (or 204) +``GET /api/taps`` recent tap history (ring buffer) +``POST /api/mode`` switch reader mode: login / monitor +============================ ============================================ + +The usual integration, for a web page that may load at any moment:: + + const r = await fetch("http://127.0.0.1:8765/api/card"); + const state = await r.json(); + if (state.claimable) { + const c = await fetch("http://127.0.0.1:8765/api/card/claim", {method: "POST"}); + if (c.ok) { + const { card } = await c.json(); + // hand card.id to your server; it decides who that is + } + } + +``claimable`` stays true for ``--tap-ttl`` seconds after the card is read - +including after the card is lifted - so a quick tap is not lost. + +Security posture (important, and deliberate): + +* Binds to ``127.0.0.1`` unless the operator explicitly opts in to more, because + an open card-event endpoint on a LAN lets anyone read who is tapping. +* Cross-origin requests are refused unless the origin is on an allow-list. + A page on another site must not be able to observe your taps. +* Optionally requires a pre-shared token (``--token`` / ``RDM_BRIDGE_TOKEN``) + for non-local callers. + +An EM4100 card id is a *username*, not a password: it is broadcast in the clear +and can be cloned. Never let holding a card id alone create a session. +""" + +from __future__ import annotations + +import json +import os +import secrets +import threading +import time +from collections import deque +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Callable, Deque, Dict, List, Optional + +from .card import EVENT_INSERTED, EVENT_REMOVED, CardEvent + +#: Reader modes. ``login`` publishes inserts only (a tap = an authentication +#: attempt). ``monitor`` publishes inserts and removals both. +MODE_LOGIN = "login" +MODE_MONITOR = "monitor" +VALID_MODES = (MODE_LOGIN, MODE_MONITOR) + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 8765 +DEFAULT_HISTORY = 50 + +#: How long a tap stays claimable. This deliberately outlives the card +#: leaving the reader: a quick tap that is lifted straight away must still be +#: claimable, otherwise a page that polls a moment later sees nothing and the +#: login silently fails. Set to 0 to keep a tap claimable until it is used or +#: replaced. +DEFAULT_TAP_TTL = 10.0 + + +class EventHub: + """Tracks reader state and fans card events out to waiting subscribers. + + This is the only place the raw card id is held. Subscribers are HTTP + connections owned by the same process, never third-party code. + + The card currently read is *latched*: it stays readable through + :meth:`current_card` after the card has left the reader, for ``tap_ttl`` + seconds from the tap. That is what makes the data "always available" to a + web page that may load or poll at any moment. + + Claiming it through :meth:`consume_card` marks it used, so a page reload or + a second tab cannot replay the same physical tap into a second login. + """ + + def __init__( + self, + history: int = DEFAULT_HISTORY, + mode: str = MODE_LOGIN, + tap_ttl: float = DEFAULT_TAP_TTL, + ): + self._lock = threading.Lock() + self._subscribers: List["deque"] = [] + self._history: Deque[CardEvent] = deque(maxlen=history) + self._last: Optional[CardEvent] = None + self._last_at: Optional[float] = None + self._mode = mode if mode in VALID_MODES else MODE_LOGIN + self._started_at = time.time() + self._counts = {EVENT_INSERTED: 0, EVENT_REMOVED: 0, "invalid_card": 0} + self._port: Optional[str] = None + self._tap_ttl = float(tap_ttl if tap_ttl else 0.0) + + # The latched tap. It deliberately outlives the card leaving the + # reader, so a page that loads after the tap can still claim it. + self._tap: Optional[CardEvent] = None + self._tap_at: Optional[float] = None + self._on_reader = False + self._consumed_at: Optional[float] = None + + # -- state ------------------------------------------------------------ + + def set_port(self, port: str) -> None: + with self._lock: + self._port = port + + @property + def mode(self) -> str: + with self._lock: + return self._mode + + def set_mode(self, mode: str) -> bool: + """Switch mode. Returns False for an unknown mode.""" + if mode not in VALID_MODES: + return False + with self._lock: + self._mode = mode + return True + + def publish(self, event: CardEvent) -> None: + """Record an event and wake every subscriber. Safe from any thread.""" + with self._lock: + self._counts[event.event] = self._counts.get(event.event, 0) + 1 + + # Presence tracking always happens, in every mode: the latched tap + # is state, not a published event. + if event.event == EVENT_INSERTED: + self._last = event + self._last_at = time.time() + # A new tap replaces any previous unclaimed one. + self._tap = event + self._tap_at = self._last_at + self._on_reader = True + self._consumed_at = None + elif event.event == EVENT_REMOVED: + # Lifting the card does NOT discard the tap: it must stay + # claimable for tap_ttl, otherwise a quick tap is lost before + # the web page gets a chance to ask for it. + self._on_reader = False + + # In login mode a tap is an authentication attempt; card removal is + # not published, so a web app cannot mistake it for a second tap. + if self._mode == MODE_LOGIN and event.event == EVENT_REMOVED: + return + + self._history.append(event) + queues = list(self._subscribers) + payload = event.to_dict() + + for queue in queues: + try: + queue.append(payload) + except Exception: + pass + + # -- latched card ----------------------------------------------------- + + def current_card(self) -> Dict[str, Any]: + """The latched tap, so a page can ask at any moment. + + ``present`` is whether the card is physically on the reader right now; + ``claimable`` is whether this tap can still be spent on a login. + """ + with self._lock: + if self._tap is None: + return { + "present": False, + "claimable": False, + "consumed": False, + "expired": False, + "age_seconds": None, + "card": None, + } + age = time.time() - (self._tap_at or 0.0) + expired = bool(self._tap_ttl) and age > self._tap_ttl + consumed = self._consumed_at is not None + return { + "present": self._on_reader, + "claimable": (not consumed) and (not expired), + "consumed": consumed, + "expired": expired, + "age_seconds": round(age, 1), + "card": self._tap.to_dict(), + } + + def consume_card(self): + """Atomically claim the latched tap for one authentication. + + Returns ``(payload, error, status)``. The tap is marked used only when + this call succeeds, so two concurrent requests cannot both win. + """ + with self._lock: + if self._tap is None: + return None, "no card read yet", 409 + + age = time.time() - (self._tap_at or 0.0) + if self._tap_ttl and age > self._tap_ttl: + return None, "card tap expired", 410 + if self._consumed_at is not None: + return None, "card already used", 409 + + self._consumed_at = time.time() + return ( + { + "card": self._tap.to_dict(), + "age_seconds": round(age, 1), + "consumed": True, + }, + None, + 200, + ) + + def subscribe(self) -> "deque": + """Register a queue that receives every subsequent event.""" + queue: Deque[Dict[str, Any]] = deque(maxlen=100) + with self._lock: + self._subscribers.append(queue) + return queue + + def unsubscribe(self, queue: "deque") -> None: + with self._lock: + try: + self._subscribers.remove(queue) + except ValueError: + pass + + # -- snapshots -------------------------------------------------------- + + def health(self) -> Dict[str, Any]: + with self._lock: + last_age = ( + round(time.time() - self._last_at, 1) if self._last_at else None + ) + return { + "status": "ok", + "port": self._port, + "mode": self._mode, + "uptime_seconds": round(time.time() - self._started_at, 1), + "subscribers": len(self._subscribers), + "counts": dict(self._counts), + "last_tap": self._last.to_dict() if self._last else None, + "last_tap_age_seconds": last_age, + "card_present": self._on_reader, + "card_consumed": self._consumed_at is not None, + "tap_ttl_seconds": self._tap_ttl or None, + } + + def last(self) -> Optional[Dict[str, Any]]: + with self._lock: + return self._last.to_dict() if self._last else None + + def taps(self, limit: int = 20) -> List[Dict[str, Any]]: + with self._lock: + items = list(self._history)[-max(1, limit):] + return [event.to_dict() for event in items] + + +class _Handler(BaseHTTPRequestHandler): + """One HTTP connection. ``server.hub`` holds the shared state.""" + + server_version = "rdm-card-reader" + protocol_version = "HTTP/1.1" + + # -- helpers ---------------------------------------------------------- + + def _cors_ok(self) -> bool: + """Refuse cross-origin reads unless the origin was allow-listed.""" + origin = self.headers.get("Origin") + if not origin: + return True # same-origin fetch or a non-browser client + allowed: List[str] = getattr(self.server, "allowed_origins", []) + if not allowed: + return False # browsers only; opt out entirely + return origin in allowed + + def _token_ok(self) -> bool: + """Check the pre-shared token for non-loopback callers.""" + required: Optional[str] = getattr(self.server, "token", None) + if not required: + return True + client = self.client_address[0] + if client in ("127.0.0.1", "::1", "localhost"): + return True + supplied = self.headers.get("X-Auth-Token") or "" + if not supplied: + query = self.path.split("?", 1)[1] if "?" in self.path else "" + for pair in query.split("&"): + if pair.startswith("token="): + supplied = pair.split("=", 1)[1] + return secrets.compare_digest(supplied, required) + + def _send_json(self, payload: Any, status: int = 200) -> None: + body = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + origin = self.headers.get("Origin") + if origin and origin in getattr(self.server, "allowed_origins", []): + self.send_header("Access-Control-Allow-Origin", origin) + self.send_header("Vary", "Origin") + self.end_headers() + self.wfile.write(body) + + def _send_no_content(self) -> None: + self.send_response(204) + self.send_header("Cache-Control", "no-store") + self.send_header("Content-Length", "0") + origin = self.headers.get("Origin") + if origin and origin in getattr(self.server, "allowed_origins", []): + self.send_header("Access-Control-Allow-Origin", origin) + self.send_header("Vary", "Origin") + self.end_headers() + + def log_message(self, fmt: str, *args: Any) -> None: + logger: Optional[Callable[[str], None]] = getattr(self.server, "logger", None) + if logger: + logger("{} - {}".format(self.client_address[0], fmt % args)) + + @property + def _hub(self) -> EventHub: + return self.server.hub # type: ignore[attr-defined] + + # -- routing ---------------------------------------------------------- + + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + path = self.path.split("?", 1)[0].rstrip("/") or "/" + + if path == "/api/health": + self._send_json(self._hub.health()) + return + + if not self._cors_ok(): + self._send_json({"error": "origin not allowed"}, status=403) + return + if not self._token_ok(): + self._send_json({"error": "missing or invalid token"}, status=401) + return + + if path == "/api/last": + payload = self._hub.last() + if payload is None: + self._send_no_content() + else: + self._send_json(payload) + return + + if path == "/api/card": + self._send_json(self._hub.current_card()) + return + + if path == "/api/taps": + limit = 20 + if "?" in self.path: + for pair in self.path.split("?", 1)[1].split("&"): + if pair.startswith("limit="): + try: + limit = int(pair.split("=", 1)[1]) + except ValueError: + pass + self._send_json({"taps": self._hub.taps(limit)}) + return + + if path == "/api/events": + self._stream_events() + return + + self._send_json({"error": "not found", "path": path}, status=404) + + def do_POST(self) -> None: # noqa: N802 + path = self.path.split("?", 1)[0].rstrip("/") or "/" + + if path == "/api/card/claim": + if not self._cors_ok(): + self._send_json({"error": "origin not allowed"}, status=403) + return + if not self._token_ok(): + self._send_json({"error": "missing or invalid token"}, status=401) + return + payload, error, status = self._hub.consume_card() + if error: + self._send_json( + {"error": error, "state": self._hub.current_card()}, + status=status, + ) + return + self._send_json(payload, status=status) + return + + if path != "/api/mode": + self._send_json({"error": "not found", "path": path}, status=404) + return + if not self._cors_ok(): + self._send_json({"error": "origin not allowed"}, status=403) + return + if not self._token_ok(): + self._send_json({"error": "missing or invalid token"}, status=401) + return + + length = int(self.headers.get("Content-Length") or 0) + raw = self.rfile.read(length) if length else b"{}" + try: + payload = json.loads(raw.decode("utf-8") or "{}") + except ValueError: + self._send_json({"error": "invalid JSON body"}, status=400) + return + + mode = str(payload.get("mode", "")) + if not self._hub.set_mode(mode): + self._send_json( + {"error": "invalid mode", "valid": list(VALID_MODES)}, status=400 + ) + return + self._send_json({"mode": mode}) + + def do_OPTIONS(self) -> None: # noqa: N802 + origin = self.headers.get("Origin") + self.send_response(204) + if origin and origin in getattr(self.server, "allowed_origins", []): + self.send_header("Access-Control-Allow-Origin", origin) + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.send_header("Access-Control-Allow-Headers", "Content-Type, X-Auth-Token") + self.send_header("Vary", "Origin") + self.send_header("Content-Length", "0") + self.end_headers() + + # -- Server-Sent Events ---------------------------------------------- + + def _stream_events(self) -> None: + """Hold the connection open and push each event as it happens.""" + queue = self._hub.subscribe() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-store") + self.send_header("Connection", "keep-alive") + origin = self.headers.get("Origin") + if origin and origin in getattr(self.server, "allowed_origins", []): + self.send_header("Access-Control-Allow-Origin", origin) + self.send_header("Vary", "Origin") + self.end_headers() + + try: + self.wfile.write(b": connected\n\n") + self.wfile.flush() + while True: + try: + payload = queue.popleft() + except IndexError: + payload = None + + if payload is not None: + message = "data: {}\n\n".format(json.dumps(payload)) + self.wfile.write(message.encode("utf-8")) + self.wfile.flush() + else: + # Comment frame keeps proxies and the browser from timing out. + self.wfile.write(b": keepalive\n\n") + self.wfile.flush() + time.sleep(0.2) + except (BrokenPipeError, ConnectionResetError): + pass + finally: + self._hub.unsubscribe(queue) + + +def create_server( + hub: EventHub, + host: str = DEFAULT_HOST, + port: int = DEFAULT_PORT, + allowed_origins: Optional[List[str]] = None, + token: Optional[str] = None, + logger: Optional[Callable[[str], None]] = None, +) -> ThreadingHTTPServer: + """Build (but do not start) the HTTP server around *hub*.""" + server = ThreadingHTTPServer((host, port), _Handler) + server.daemon_threads = True + server.hub = hub # type: ignore[attr-defined] + server.allowed_origins = list(allowed_origins or []) # type: ignore[attr-defined] + server.token = token or os.environ.get("RDM_BRIDGE_TOKEN") or None # type: ignore[attr-defined] + server.logger = logger # type: ignore[attr-defined] + return server + + +def is_loopback(host: str) -> bool: + """True for hosts that are not reachable off-box.""" + return host in ("127.0.0.1", "::1", "localhost") diff --git a/src/rdm_card_reader/card.py b/src/rdm_card_reader/card.py new file mode 100644 index 0000000..437777f --- /dev/null +++ b/src/rdm_card_reader/card.py @@ -0,0 +1,135 @@ +"""Card data model, protocol constants and output formatting. + +Nothing in this module touches hardware or the serial port, which keeps the +data model easy to unit test. +""" + +from __future__ import annotations + +import json +import time +from collections import namedtuple +from dataclasses import dataclass +from typing import Any, Dict + +try: # pragma: no cover - depends on the environment + from rdm6300 import CardData +except ImportError: # pragma: no cover - fallback so the model stays importable + CardData = namedtuple("CardData", ["value", "checksum", "type", "is_valid"]) + +#: Serial devices tried in order. ``/dev/ttyS0`` first matches the Raspberry Pi +#: default, the others cover USB / GPIO UART dongles. +CANDIDATE_PORTS = ( + "/dev/ttyS0", + "/dev/ttyAMA0", + "/dev/ttyUSB0", + "/dev/ttyACM0", +) + +#: A special card id used by the WMT workstation app to open its config screen. +#: Reported as a note only - this package never acts on it. +CONFIG_CARD_VALUE = 12886709 + +#: Seconds without a read before a card is considered removed. Set to ``None`` +#: to disable removal detection entirely. +DEFAULT_HEARTBEAT_INTERVAL = 0.5 + +BAR = "-" * 58 + +EVENT_INSERTED = "card_inserted" +EVENT_REMOVED = "card_removed" +EVENT_INVALID = "invalid_card" + + +def now_stamp() -> str: + """Local timestamp in the same format the workstation app logs.""" + return time.strftime("%Y-%m-%d %H:%M:%S") + + +@dataclass(frozen=True) +class CardEvent: + """A single card read event, decoupled from the reader library.""" + + event: str + timestamp: str + value: int + checksum: int + card_type: int + is_valid: bool + port: str + + # -- constructors ----------------------------------------------------- + + @classmethod + def from_card(cls, event: str, card: Any, port: str) -> "CardEvent": + """Build an event from an ``rdm6300.CardData`` (or anything alike).""" + return cls( + event=event, + timestamp=now_stamp(), + value=int(card.value), + checksum=int(card.checksum), + card_type=int(card.type), + is_valid=bool(card.is_valid), + port=str(port), + ) + + # -- derived values --------------------------------------------------- + + @property + def hex_value(self) -> str: + """Card id as an 8 digit hex string, e.g. ``0x00C4A0B5``.""" + return "0x{:08X}".format(self.value) + + @property + def is_config_card(self) -> bool: + return self.value == CONFIG_CARD_VALUE + + # -- serialisation ---------------------------------------------------- + + def to_dict(self) -> Dict[str, Any]: + """Machine-readable form, used by ``--json``.""" + return { + "event": self.event, + "timestamp": self.timestamp, + "id": self.value, + "hex": self.hex_value, + "type": self.card_type, + "checksum": self.checksum, + "is_valid": self.is_valid, + "port": self.port, + } + + def to_json(self) -> str: + return json.dumps(self.to_dict()) + + +def format_event(event: CardEvent) -> str: + """Render an event as the human-readable block printed on stdout.""" + if event.event == EVENT_INSERTED: + lines = [ + "", + BAR, + " CARD INSERTED - {}".format(event.timestamp), + " id : {} ({})".format(event.value, event.hex_value), + " type : 0x{:02X}".format(event.card_type), + " checksum : {}".format(event.checksum), + " valid : {}".format(event.is_valid), + " port : {}".format(event.port), + ] + if event.is_config_card: + lines.append( + " note : this is the config card (id {})".format( + CONFIG_CARD_VALUE + ) + ) + lines.append(BAR) + return "\n".join(lines) + + if event.event == EVENT_REMOVED: + return " CARD REMOVED - {} id: {} ({})".format( + event.timestamp, event.value, event.hex_value + ) + + return " INVALID CARD - {} raw: {} (checksum mismatch)".format( + event.timestamp, event.hex_value + ) diff --git a/src/rdm_card_reader/cli.py b/src/rdm_card_reader/cli.py new file mode 100644 index 0000000..e5dcc6c --- /dev/null +++ b/src/rdm_card_reader/cli.py @@ -0,0 +1,285 @@ +"""Command line interface for the RDM card reader.""" + +from __future__ import annotations + +import argparse +import sys +import threading +from typing import Optional, Sequence + +from . import __version__ +from .bridge import ( + DEFAULT_HOST, + DEFAULT_PORT, + DEFAULT_TAP_TTL, + MODE_LOGIN, + VALID_MODES, + EventHub, + create_server, +) +from .card import ( + BAR, + CANDIDATE_PORTS, + DEFAULT_HEARTBEAT_INTERVAL, + CardEvent, + format_event, +) +from .demo import run_demo +from .ports import AmbiguousPort, PortNotFound, discover, resolve +from .reader import available, explain_error, list_ports, open_reader + +EXIT_OK = 0 +EXIT_NO_PORT = 1 +EXIT_NO_LIBRARY = 2 +EXIT_PORT_AMBIGUOUS = 3 + + +def _make_printer(as_json: bool): + """Return a callback that writes one event to stdout.""" + + def print_event(event: CardEvent) -> None: + if as_json: + print(event.to_json(), flush=True) + else: + print(format_event(event), flush=True) + + return print_event + + +def _show_ports(verbose: bool = False) -> int: + info = list_ports() + print("Candidate serial ports (tried in this order):") + for port, exists in info["candidates"]: + print(" {} {}".format("OK " if exists else "---", port)) + + ports = discover() + if ports: + print("\nIdentified serial ports (best candidate first):") + for entry in sorted(ports, key=lambda p: (-p.score(), p.device)): + print(" [score {:>3}] {}".format(entry.score(), entry.describe())) + if verbose: + stable = entry.stable_id or "(none)" + print(" by-id : {}".format(stable)) + print(" by-path : {}".format(entry.by_path or "(none)")) + print(" location : {}".format(entry.location or "(unknown)")) + print(" usb id : {}".format(entry.usb_id or "(not USB)")) + print(" driver : {}".format(entry.product or "(unknown)")) + else: + print("\n(no serial ports identified)") + + print("\nTip: pass --port with a serial number or by-id path so the same") + print(" physical adapter is always used, e.g.") + print(" --port /dev/serial/by-id/usb-1a86_USB_Serial-if00-port0") + return EXIT_OK + + +def _resolve_port(args) -> str: + """Work out which serial port to open, or exit with a clear reason.""" + selectors = [args.port] if args.port else None + try: + device, reason = resolve( + selectors=selectors, + fallback_ports=[args.port] if args.port else list(CANDIDATE_PORTS), + autodetect=args.autodetect, + ) + except AmbiguousPort as exc: + print("x {}".format(exc), file=sys.stderr) + print(" Refusing to guess which reader is the right one.", file=sys.stderr) + sys.exit(EXIT_PORT_AMBIGUOUS) + except PortNotFound as exc: + print("x {}".format(exc), file=sys.stderr) + sys.exit(EXIT_NO_PORT) + + print("-> Using {} ({})".format(device, reason), flush=True) + return device + + +def _run_bridge(args) -> int: + """Serve what the reader sees over HTTP for a web app to consume.""" + if not available(): + print("ERROR: the 'rdm6300' library is not installed.", file=sys.stderr) + print(" Install it with: pip install rdm6300", file=sys.stderr) + print(" Or run with --demo to see the output format.", file=sys.stderr) + return EXIT_NO_LIBRARY + + if not args.allow_remote and args.host not in ("127.0.0.1", "::1", "localhost"): + print("ERROR: refusing to bind {} without --allow-remote.".format(args.host), + file=sys.stderr) + print(" A card-event endpoint on the LAN exposes who taps where.", + file=sys.stderr) + return 1 + + device = _resolve_port(args) + + hub = EventHub(mode=args.mode, tap_ttl=args.tap_ttl) + hub.set_port(device) + + def emit(event: CardEvent) -> None: + hub.publish(event) + if args.verbose: + print(format_event(event), flush=True) + + def report_error(port: str, message: str) -> None: + print(" x {}: {}".format(port, explain_error(port, message)), flush=True) + + reader = open_reader(emit, ports=[device], + heartbeat_interval=args.heartbeat, + on_error=report_error) + if reader is None: + print("x Could not open {}. Is another process using it?".format(device), + file=sys.stderr) + return EXIT_NO_PORT + + server = create_server( + hub, + host=args.host, + port=args.bridge_port, + allowed_origins=args.allow_origin, + token=args.token, + logger=print if args.verbose else None, + ) + + server_thread = threading.Thread(target=server.serve_forever, daemon=True, + name="bridge-http") + server_thread.start() + + print(" OK reader on {} ; bridge on http://{}:{} (mode: {})".format( + device, args.host, args.bridge_port, args.mode)) + print(" GET /api/card <- the card on the reader right now") + print(" POST /api/card/claim (single use, for one login)") + print(" GET /api/events (SSE stream)") + print(" GET /api/last /api/health") + if args.token: + print(" token required for non-local callers") + if not args.allow_origin: + print(" note: cross-origin browser requests are refused") + print(" pass --allow-origin https://your.app to permit one") + print(" Ctrl+C to stop.", flush=True) + + # The reader loop blocks this thread; the HTTP server runs alongside it. + try: + reader.start() + except KeyboardInterrupt: + print("\nStopped by user.") + finally: + try: + reader.close() + except Exception: + pass + server.shutdown() + server.server_close() + return EXIT_OK + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="rdm-card-reader", + description="Read RFID cards (RDM6300 / EM4100) and print the card " + "inserted and card removed output.", + ) + parser.add_argument("--version", action="version", + version="%(prog)s {}".format(__version__)) + parser.add_argument("--port", help="serial port: device path, by-id symlink, " + "USB serial number or vid:pid") + parser.add_argument("--autodetect", action="store_true", + help="pick the only USB serial adapter; refuse if unsure") + parser.add_argument("--heartbeat", type=float, + default=DEFAULT_HEARTBEAT_INTERVAL, + help="seconds without a read before the card is reported " + "as removed (default: %(default)s)") + parser.add_argument("--json", action="store_true", + help="print one JSON object per event") + parser.add_argument("--list-ports", action="store_true", + help="list candidate and identified ports, then exit") + parser.add_argument("--verbose-ports", action="store_true", + help="with --list-ports, show by-id / by-path detail") + parser.add_argument("--demo", action="store_true", + help="simulate card reads (needs no hardware or library)") + parser.add_argument("--demo-count", type=int, default=3, + help="number of simulated reads in --demo (default: %(default)s)") + + bridge = parser.add_argument_group("HTTP bridge (--serve)") + bridge.add_argument("--serve", action="store_true", + help="serve card events over HTTP for a web app") + bridge.add_argument("--host", default=DEFAULT_HOST, + help="bind address (default: %(default)s, loopback only)") + bridge.add_argument("--bridge-port", type=int, default=DEFAULT_PORT, + help="HTTP port (default: %(default)s)") + bridge.add_argument("--mode", default=MODE_LOGIN, choices=list(VALID_MODES), + help="login: publish taps only; monitor: taps and removals " + "(default: %(default)s)") + bridge.add_argument("--tap-ttl", type=float, default=DEFAULT_TAP_TTL, + help="seconds an unclaimed tap stays claimable; 0 keeps it " + "claimable while the card is present (default: %(default)s)") + bridge.add_argument("--allow-origin", action="append", metavar="ORIGIN", + help="permit this browser origin to call the API; repeatable") + bridge.add_argument("--token", + help="pre-shared token required from non-local callers " + "(or set RDM_BRIDGE_TOKEN)") + bridge.add_argument("--allow-remote", action="store_true", + help="permit binding a non-loopback address") + bridge.add_argument("--verbose", action="store_true", + help="also print each event and each HTTP request") + return parser + + +def main(argv: Optional[Sequence[str]] = None) -> int: + args = build_parser().parse_args(argv) + + if args.list_ports: + return _show_ports(verbose=args.verbose_ports) + + if args.demo: + return run_demo(count=args.demo_count, heartbeat=args.heartbeat, + as_json=args.json) + + if args.serve: + return _run_bridge(args) + + if not available(): + print("ERROR: the 'rdm6300' library is not installed.", file=sys.stderr) + print(" Install it with: pip install rdm6300", file=sys.stderr) + print(" Or run with --demo to see the output format.", file=sys.stderr) + return EXIT_NO_LIBRARY + + print_event = _make_printer(args.json) + ports = [args.port] if args.port else list(CANDIDATE_PORTS) + + def report_error(port: str, message: str) -> None: + print(" x {}: {}".format(port, explain_error(port, message)), flush=True) + + for port in ports: + print("-> Opening {} ...".format(port), flush=True) + + reader = open_reader(print_event, ports=ports, + heartbeat_interval=args.heartbeat, + on_error=report_error) + + if reader is None: + print("\nx Could not open any serial port. Check:", file=sys.stderr) + print(" 1. the reader is connected (rdm-card-reader --list-ports)", + file=sys.stderr) + print(" 2. UART is enabled: sudo raspi-config -> Interface Options -> Serial", + file=sys.stderr) + print(" 3. the user is in dialout: sudo usermod -aG dialout $USER", + file=sys.stderr) + return EXIT_NO_PORT + + print(" OK reader ready on {} - waiting for cards (Ctrl+C to stop)".format( + reader.port), flush=True) + + try: + reader.start() # blocks; events arrive through the callback + except KeyboardInterrupt: + print("\nStopped by user.") + finally: + try: + reader.close() + except Exception: + pass + + return EXIT_OK + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main()) diff --git a/src/rdm_card_reader/demo.py b/src/rdm_card_reader/demo.py new file mode 100644 index 0000000..14e5bf7 --- /dev/null +++ b/src/rdm_card_reader/demo.py @@ -0,0 +1,49 @@ +"""Simulated card reads, so the output can be checked without hardware.""" + +from __future__ import annotations + +import random +import time + +from .card import ( + DEFAULT_HEARTBEAT_INTERVAL, + CardData, + CardEvent, + format_event, +) + +# Card ids with the checksum the real protocol would produce +# (XOR of the type byte and the four id bytes), so the output looks genuine. +DEMO_CARDS = ( + CardData(value=0x00C4A0B5, checksum=0xDE, type=0x0F, is_valid=True), + CardData(value=0x1A2B3C4D, checksum=0x4F, type=0x0F, is_valid=True), + CardData(value=0xDEADBEEF, checksum=0x00, type=0x0F, is_valid=False), +) + + +def _emit(event: CardEvent, as_json: bool) -> None: + print(event.to_json() if as_json else format_event(event), flush=True) + + +def run_demo( + count: int = 3, + heartbeat: float = DEFAULT_HEARTBEAT_INTERVAL, + as_json: bool = False, + seed: int = 1234, +) -> int: + """Print *count* simulated card read cycles. Returns a process exit code.""" + rng = random.Random(seed) + print("Demo mode - no hardware is used, values are simulated.\n", flush=True) + + for _ in range(max(1, count)): + card = rng.choice(DEMO_CARDS) + event_name = "card_inserted" if card.is_valid else "invalid_card" + _emit(CardEvent.from_card(event_name, card, "demo"), as_json) + + time.sleep(max(0.1, heartbeat)) + if card.is_valid: + _emit(CardEvent.from_card("card_removed", card, "demo"), as_json) + time.sleep(0.3) + + print("\nDemo finished.", flush=True) + return 0 diff --git a/src/rdm_card_reader/ports.py b/src/rdm_card_reader/ports.py new file mode 100644 index 0000000..731e18d --- /dev/null +++ b/src/rdm_card_reader/ports.py @@ -0,0 +1,397 @@ +"""Identify serial ports reliably, so the reader does not grab the wrong ttyUSB. + +A bare ``/dev/ttyUSB0`` is not a stable identity: unplug two adapters and the +numbers swap. This module answers "is this really the RFID reader?" using the +things that *are* stable: + +* the USB serial number (``/dev/serial/by-id``) +* the physical USB port (``/dev/serial/by-path``) +* the USB vendor:product id and the adapter description + +Use :func:`resolve` to turn a user-supplied selector into exactly one port, +refusing to guess when the answer is ambiguous. +""" + +from __future__ import annotations + +import glob +import os +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Sequence, Tuple + +#: USB-serial bridge chips commonly found on RDM6300 / Arduino adapters. +#: Mapping of ``vid:pid`` -> friendly name. Used for scoring, not required. +KNOWN_USB_SERIAL_CHIPS: Dict[Tuple[int, int], str] = { + (0x1A86, 0x7523): "CH340", + (0x1A86, 0x5523): "CH341", + (0x1A86, 0x55D4): "CH9102", + (0x0403, 0x6001): "FT232R", + (0x0403, 0x6015): "FT231X", + (0x10C4, 0xEA60): "CP2102", + (0x10C4, 0xEA70): "CP2105", + (0x067B, 0x2303): "PL2303", + (0x2341, 0x0043): "Arduino Uno", + (0x1A86, 0x55D3): "CH343", +} + +#: Description fragments that suggest a generic UART bridge. +_SERIAL_HINTS = ("serial", "uart", "usb-serial", "usbserial", "ttyusb", "ttyacm") + + +class PortError(RuntimeError): + """Base class for port selection problems.""" + + +class PortNotFound(PortError): + """No port matched the requested selector.""" + + +class AmbiguousPort(PortError): + """More than one port matched; picking one silently would be unsafe.""" + + +@dataclass +class PortInfo: + """Everything stable we can learn about one serial port.""" + + device: str + description: str = "" + hwid: str = "" + vid: Optional[int] = None + pid: Optional[int] = None + serial_number: Optional[str] = None + manufacturer: Optional[str] = None + product: Optional[str] = None + location: Optional[str] = None + aliases: List[str] = field(default_factory=list) + + # -- derived ---------------------------------------------------------- + + @property + def is_usb(self) -> bool: + """True when this port is a USB device (as opposed to an on-board UART).""" + return self.vid is not None or "usb" in self.device.lower() + + @property + def usb_id(self) -> Optional[str]: + """``"1a86:7523"`` when both ids are known.""" + if self.vid is None or self.pid is None: + return None + return "{:04x}:{:04x}".format(self.vid, self.pid) + + @property + def chip(self) -> Optional[str]: + """Friendly name of the USB-serial bridge, when recognised.""" + if self.vid is None or self.pid is None: + return None + return KNOWN_USB_SERIAL_CHIPS.get((self.vid, self.pid)) + + @property + def stable_id(self) -> Optional[str]: + """A ``/dev/serial/by-id`` path, when the kernel provides one. + + This is the value to persist in a config file - it survives replugging + and reordering. + """ + for alias in self.aliases: + if "/by-id/" in alias: + return alias + return None + + @property + def by_path(self) -> Optional[str]: + """A ``/dev/serial/by-path`` path - stable per physical USB socket.""" + for alias in self.aliases: + if "/by-path/" in alias: + return alias + return None + + def score(self) -> int: + """Likely-to-be-an-RFID-adapter score. Higher is better. + + Deliberately conservative: a plain on-board ``/dev/ttyS0`` scores low + so that a real USB adapter wins the autodetect race. + """ + value = 0 + if self.chip: + value += 60 + elif self.is_usb: + value += 40 + if self.serial_number: + value += 20 # uniquely identifiable + if self.stable_id: + value += 10 + if any(hint in self.description.lower() for hint in _SERIAL_HINTS): + value += 10 + if self.device.startswith("/dev/serial"): + value += 15 # kernel-managed alias + if self.device.startswith("/dev/ttyS"): + value += 5 # on-board UART, still valid but less specific + return value + + def _haystacks(self) -> List[str]: + parts = [ + self.device, + self.description, + self.hwid, + self.manufacturer or "", + self.product or "", + self.serial_number or "", + self.usb_id or "", + ] + parts.extend(self.aliases) + return [p.lower() for p in parts if p] + + def matches(self, selector: str) -> bool: + """True when *selector* identifies this port. + + Accepts a device path, a ``by-id``/``by-path`` symlink, a serial + number, ``vid:pid``, a manufacturer/product fragment or a chip name. + """ + needle = selector.strip().lower() + if not needle: + return False + if self.device.lower() == needle: + return True + if self.serial_number and self.serial_number.lower() == needle: + return True + if self.usb_id and self.usb_id == needle: + return True + if self.chip and self.chip.lower() == needle: + return True + return any(needle in hay for hay in self._haystacks()) + + def to_dict(self) -> Dict[str, Any]: + return { + "device": self.device, + "description": self.description, + "usb_id": self.usb_id, + "chip": self.chip, + "serial_number": self.serial_number, + "manufacturer": self.manufacturer, + "product": self.product, + "location": self.location, + "stable_id": self.stable_id, + "by_path": self.by_path, + "is_usb": self.is_usb, + "score": self.score(), + } + + def describe(self) -> str: + """One-line summary for ``--list-ports``.""" + bits = [self.device] + if self.chip: + bits.append(self.chip) + elif self.description: + bits.append(self.description) + if self.usb_id: + bits.append(self.usb_id) + if self.serial_number: + bits.append("S/N {}".format(self.serial_number)) + return " :: ".join(bits) + + +# --------------------------------------------------------------------------- +# Discovery +# --------------------------------------------------------------------------- + +def _aliases_for(device: str) -> List[str]: + """Collect ``/dev/serial/by-id`` and ``by-path`` symlinks pointing at *device*.""" + found: List[str] = [] + for folder in ("by-id", "by-path"): + pattern = "/dev/serial/{}/**".format(folder) + for link in sorted(glob.glob(pattern, recursive=True)): + try: + if os.path.realpath(link) == os.path.realpath(device): + found.append(link) + except OSError: + continue + return found + + +def _info_from_sysfs(device: str) -> PortInfo: + """Fallback identification straight from sysfs (no pyserial needed).""" + info = PortInfo(device=device) + try: + info.aliases = _aliases_for(device) + except OSError: + pass + + name = os.path.basename(device) + base = "/sys/class/tty/{}/device".format(name) + if not os.path.exists(base): + return info + + def _read(relpath: str) -> Optional[str]: + try: + with open(os.path.join(base, relpath), "r") as handle: + return handle.read().strip() + except OSError: + return None + + vid = _read("../idVendor") + pid = _read("../idProduct") + if vid: + try: + info.vid = int(vid, 16) + except ValueError: + pass + if pid: + try: + info.pid = int(pid, 16) + except ValueError: + pass + + info.serial_number = _read("../serial") + info.manufacturer = _read("../manufacturer") + info.product = _read("../product") + interface = _read("interface") + if interface: + info.description = interface + return info + + +def discover() -> List[PortInfo]: + """List every serial port with its stable identity. + + Prefers pyserial's enumeration (richer metadata) and falls back to scanning + ``/dev`` plus sysfs, so it still works when pyserial is absent. + """ + ports: List[PortInfo] = [] + + try: + from serial.tools import list_ports as serial_list_ports + + for entry in serial_list_ports.comports(): + info = PortInfo( + device=entry.device, + description=entry.description or "", + hwid=entry.hwid or "", + vid=getattr(entry, "vid", None), + pid=getattr(entry, "pid", None), + serial_number=getattr(entry, "serial_number", None), + manufacturer=getattr(entry, "manufacturer", None), + product=getattr(entry, "product", None), + location=getattr(entry, "location", None), + ) + info.aliases = _aliases_for(entry.device) + ports.append(info) + + if ports: + return ports + except Exception: + pass + + # Fallback: scan /dev for tty devices, enriching from sysfs. + candidates = sorted(glob.glob("/dev/ttyUSB*")) + sorted(glob.glob("/dev/ttyACM*")) + candidates += sorted(glob.glob("/dev/ttyS*")) + sorted(glob.glob("/dev/ttyAMA*")) + candidates += sorted(glob.glob("/dev/serial/by-id/*")) + + seen = set() + for device in candidates: + real = os.path.realpath(device) + if real in seen: + continue + seen.add(real) + ports.append(_info_from_sysfs(device)) + + return ports + + +def best_usb_candidates(ports: Sequence[PortInfo]) -> List[PortInfo]: + """USB-serial ports sorted most-likely-first, duplicates by identity removed.""" + usb = [p for p in ports if p.is_usb] + usb.sort(key=lambda p: (-p.score(), p.device)) + + unique: List[PortInfo] = [] + identities = set() + for port in usb: + identity = port.stable_id or port.usb_id or port.device + if port.serial_number: + identity = "{}|{}".format(port.usb_id, port.serial_number) + if identity in identities: + continue + identities.add(identity) + unique.append(port) + return unique + + +# --------------------------------------------------------------------------- +# Resolution +# --------------------------------------------------------------------------- + +def resolve( + selectors: Optional[Sequence[str]] = None, + fallback_ports: Optional[Sequence[str]] = None, + autodetect: bool = False, +) -> Tuple[str, str]: + """Decide which port to open. + + Returns ``(device, reason)`` where *reason* explains the choice so the + caller can log it - being explicit about *why* a port was picked is the + whole point of this module. + + Raises :class:`PortNotFound` or :class:`AmbiguousPort` rather than + guessing. + """ + ports = discover() + + # 1. Explicit selectors always win. + if selectors: + matched: List[PortInfo] = [] + for selector in selectors: + for port in ports: + if port.matches(selector) and port not in matched: + matched.append(port) + + if not matched: + # The selector may be a raw path pyserial cannot enumerate. + for selector in selectors: + if os.path.exists(selector): + return selector, "explicit path {}".format(selector) + available = ", ".join(p.device for p in ports) or "none detected" + raise PortNotFound( + "no port matched {}. Detected ports: {}".format( + ", ".join(repr(s) for s in selectors), available + ) + ) + + if len(matched) > 1: + raise AmbiguousPort( + "{} ports matched: {}. Use --port with a serial number or " + "by-id path, or --list-ports to see them.".format( + len(matched), ", ".join(p.describe() for p in matched) + ) + ) + + port = matched[0] + return port.device, "matched {} ({})".format( + port.stable_id or port.usb_id or "selector", port.describe() + ) + + # 2. Autodetect: only trust a single unambiguous USB adapter. + if autodetect: + candidates = best_usb_candidates(ports) + if len(candidates) == 1: + port = candidates[0] + return port.device, "autodetected {} (score {})".format( + port.describe(), port.score() + ) + if len(candidates) > 1: + raise AmbiguousPort( + "{} USB serial ports found; refusing to guess. Use --port or " + "set a serial_number in the config. Candidates: {}".format( + len(candidates), ", ".join(p.describe() for p in candidates) + ) + ) + + # 3. Fall back to the fixed order, which is what the workstation app used. + for device in list(fallback_ports or []): + if os.path.exists(device): + return device, "first available of the default order" + + raise PortNotFound( + "no serial port found. Detected: {}".format( + ", ".join(p.describe() for p in ports) or "none" + ) + ) diff --git a/src/rdm_card_reader/py.typed b/src/rdm_card_reader/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/rdm_card_reader/reader.py b/src/rdm_card_reader/reader.py new file mode 100644 index 0000000..97e8ff1 --- /dev/null +++ b/src/rdm_card_reader/reader.py @@ -0,0 +1,111 @@ +"""Serial-port handling: open the RDM6300 and turn reads into events. + +The reader library is a hard dependency at runtime but is imported lazily so +that ``--demo`` and the unit tests keep working without it installed. +""" + +from __future__ import annotations + +import os +from typing import Callable, Dict, List, Optional, Sequence + +from .card import CANDIDATE_PORTS, DEFAULT_HEARTBEAT_INTERVAL, CardEvent + +try: # pragma: no cover - depends on the environment + import rdm6300 +except ImportError: # pragma: no cover + rdm6300 = None # type: ignore[assignment] + +#: Called once per card event. Keep it fast - it runs on the reader thread. +EventCallback = Callable[[CardEvent], None] + + +class ReaderUnavailable(RuntimeError): + """Raised when the ``rdm6300`` library is not installed.""" + + +def available() -> bool: + """True when the ``rdm6300`` library can be imported.""" + return rdm6300 is not None + + +def _make_reader_class(callback: EventCallback): + """Build a ``BaseReader`` subclass that reports events via *callback*.""" + if rdm6300 is None: + raise ReaderUnavailable( + "the 'rdm6300' library is not installed (pip install rdm6300)" + ) + + class CardReader(rdm6300.BaseReader): # type: ignore[misc, valid-type] + """Fires *callback* for each insert, removal and checksum failure.""" + + def card_inserted(self, card) -> None: + callback(CardEvent.from_card("card_inserted", card, self.port)) + + def card_removed(self, card) -> None: + if card is None: + return + callback(CardEvent.from_card("card_removed", card, self.port)) + + def invalid_card(self, card) -> None: + callback(CardEvent.from_card("invalid_card", card, self.port)) + + return CardReader + + +def open_reader( + callback: EventCallback, + ports: Optional[Sequence[str]] = None, + heartbeat_interval: Optional[float] = DEFAULT_HEARTBEAT_INTERVAL, + on_error: Optional[Callable[[str, str], None]] = None, +): + """Open the first usable port and return the reader (not yet started). + + Returns ``None`` when no port could be opened. *on_error* receives + ``(port, message)`` for each failed attempt so the caller can print or log + the reason. + """ + candidates = list(ports) if ports else list(CANDIDATE_PORTS) + ReaderClass = _make_reader_class(callback) + + for port in candidates: + try: + reader = ReaderClass(port, heartbeat_interval=heartbeat_interval) + except Exception as exc: # port missing, permission denied, busy, ... + if on_error is not None: + on_error(port, str(exc)) + continue + return reader + + return None + + +def explain_error(port: str, message: str) -> str: + """Turn a pyserial error into an actionable hint.""" + lowered = message.lower() + if "permission denied" in lowered: + return ( + "permission denied - add the user to the 'dialout' group: " + "sudo usermod -aG dialout $USER (then log in again)" + ) + if "no such file" in lowered or "could not open port" in lowered: + return "device not found" + return message + + +def list_ports() -> Dict[str, List]: + """Report candidate ports and whatever pyserial can enumerate.""" + candidates = [(port, os.path.exists(port)) for port in CANDIDATE_PORTS] + + detected: List[str] = [] + try: + from serial.tools import list_ports as serial_list_ports + + detected = [ + "{} ({})".format(info.device, info.description) + for info in serial_list_ports.comports() + ] + except Exception: + detected = [] + + return {"candidates": candidates, "detected": detected} diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..dac5d19 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,8 @@ +import os +import sys + +SRC = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" +) +if SRC not in sys.path: + sys.path.insert(0, SRC) diff --git a/tests/test_bridge.py b/tests/test_bridge.py new file mode 100644 index 0000000..7fdfa9f --- /dev/null +++ b/tests/test_bridge.py @@ -0,0 +1,501 @@ +"""Tests for the HTTP bridge. + +Two things matter most here: + +1. The bridge publishes card events without claiming to authenticate anyone. +2. A page on another origin must not be able to read your taps. +""" + +import json +import os +import sys +import threading +import time +import urllib.error +import urllib.request + +SRC = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" +) +if SRC not in sys.path: + sys.path.insert(0, SRC) + +import pytest # noqa: E402 + +from rdm_card_reader import bridge # noqa: E402 +from rdm_card_reader.bridge import ( # noqa: E402 + MODE_LOGIN, + MODE_MONITOR, + EventHub, + create_server, +) +from rdm_card_reader.card import CardEvent # noqa: E402 + + +def make_event(event="card_inserted", value=12886197): + return CardEvent( + event=event, + timestamp="2026-10-01 12:00:00", + value=value, + checksum=222, + card_type=15, + is_valid=True, + port="/dev/ttyUSB0", + ) + + +# -- hub -------------------------------------------------------------------- + +def test_hub_publishes_to_subscriber(): + hub = EventHub() + queue = hub.subscribe() + hub.publish(make_event()) + assert len(queue) == 1 + assert queue[0]["id"] == 12886197 + + +def test_hub_login_mode_hides_removals(): + """Removal must not look like a second authentication attempt.""" + hub = EventHub(mode=MODE_LOGIN) + queue = hub.subscribe() + hub.publish(make_event("card_removed")) + assert len(queue) == 0 + + +def test_hub_monitor_mode_publishes_removals(): + hub = EventHub(mode=MODE_MONITOR) + queue = hub.subscribe() + hub.publish(make_event("card_removed")) + assert len(queue) == 1 + + +def test_hub_unsubscribe_stops_delivery(): + hub = EventHub() + queue = hub.subscribe() + hub.unsubscribe(queue) + hub.publish(make_event()) + assert len(queue) == 0 + + +def test_hub_tracks_last_tap_only_on_insert(): + hub = EventHub(mode=MODE_MONITOR) + hub.publish(make_event("card_inserted", value=111)) + hub.publish(make_event("card_removed", value=111)) + assert hub.last()["id"] == 111 + + +def test_hub_history_is_bounded(): + hub = EventHub(history=3) + for value in range(10): + hub.publish(make_event(value=value)) + assert len(hub.taps(limit=99)) == 3 + + +def test_hub_health_reports_state(): + hub = EventHub() + hub.set_port("/dev/ttyUSB0") + hub.publish(make_event()) + health = hub.health() + assert health["port"] == "/dev/ttyUSB0" + assert health["counts"]["card_inserted"] == 1 + assert health["last_tap"]["id"] == 12886197 + assert health["last_tap_age_seconds"] is not None + + +def test_hub_rejects_unknown_mode(): + hub = EventHub() + assert hub.set_mode("bogus") is False + assert hub.mode == MODE_LOGIN + + +def test_hub_accepts_valid_mode_switch(): + hub = EventHub() + assert hub.set_mode(MODE_MONITOR) is True + assert hub.mode == MODE_MONITOR + + +# -- HTTP ------------------------------------------------------------------- + +@pytest.fixture +def server(): + """A live bridge on an ephemeral port, torn down after each test.""" + hub = EventHub() + hub.set_port("/dev/ttyUSB0") + srv = create_server(hub, host="127.0.0.1", port=0) + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + yield srv, hub + srv.shutdown() + srv.server_close() + + +def base_url(srv): + host, port = srv.server_address[:2] + return "http://{}:{}".format(host, port) + + +def get(srv, path, headers=None): + request = urllib.request.Request(base_url(srv) + path, headers=headers or {}) + with urllib.request.urlopen(request, timeout=5) as response: + return response.status, response.headers, response.read() + + +def test_health_endpoint(server): + srv, hub = server + status, _, body = get(srv, "/api/health") + assert status == 200 + payload = json.loads(body) + assert payload["status"] == "ok" + assert payload["port"] == "/dev/ttyUSB0" + assert payload["mode"] == MODE_LOGIN + + +def test_last_returns_204_when_nothing_read(server): + srv, _ = server + status, _, _ = get(srv, "/api/last") + assert status == 204 + + +def test_last_returns_the_tap(server): + srv, hub = server + hub.publish(make_event()) + status, _, body = get(srv, "/api/last") + assert status == 200 + assert json.loads(body)["id"] == 12886197 + + +def test_taps_endpoint_honours_limit(server): + srv, hub = server + for value in range(5): + hub.publish(make_event(value=value)) + _, _, body = get(srv, "/api/taps?limit=2") + assert len(json.loads(body)["taps"]) == 2 + + +def test_unknown_path_is_404(server): + srv, _ = server + with pytest.raises(urllib.error.HTTPError) as excinfo: + get(srv, "/api/nope") + assert excinfo.value.code == 404 + + +def test_health_available_without_origin(server): + """Monitoring must work for a plain client with no Origin header.""" + srv, _ = server + status, _, _ = get(srv, "/api/health") + assert status == 200 + + +# -- origin policy ---------------------------------------------------------- + +def test_cross_origin_refused_when_no_allowlist(server): + """A third-party page must not be able to read card ids.""" + srv, _ = server + with pytest.raises(urllib.error.HTTPError) as excinfo: + get(srv, "/api/last", headers={"Origin": "https://evil.example"}) + assert excinfo.value.code == 403 + + +def test_allowlisted_origin_permitted_and_echoed(): + hub = EventHub() + srv = create_server(hub, host="127.0.0.1", port=0, + allowed_origins=["https://kiosk.example"]) + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + try: + hub.publish(make_event()) + status, headers, body = get( + srv, "/api/last", headers={"Origin": "https://kiosk.example"} + ) + assert status == 200 + assert headers.get("Access-Control-Allow-Origin") == "https://kiosk.example" + assert json.loads(body)["id"] == 12886197 + finally: + srv.shutdown() + srv.server_close() + + +def test_unlisted_origin_still_refused_when_allowlist_set(): + hub = EventHub() + srv = create_server(hub, host="127.0.0.1", port=0, + allowed_origins=["https://kiosk.example"]) + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + try: + with pytest.raises(urllib.error.HTTPError) as excinfo: + get(srv, "/api/last", headers={"Origin": "https://other.example"}) + assert excinfo.value.code == 403 + finally: + srv.shutdown() + srv.server_close() + + +def test_token_not_required_from_loopback(): + hub = EventHub() + srv = create_server(hub, host="127.0.0.1", port=0, token="sekrit") + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + try: + status, _, _ = get(srv, "/api/health") + assert status == 200 + finally: + srv.shutdown() + srv.server_close() + + +# -- mode switching --------------------------------------------------------- + +def post(srv, path, payload, headers=None): + data = json.dumps(payload).encode() + request = urllib.request.Request( + base_url(srv) + path, data=data, method="POST", + headers={"Content-Type": "application/json", **(headers or {})}, + ) + with urllib.request.urlopen(request, timeout=5) as response: + return response.status, response.read() + + +def test_mode_can_be_switched(server): + srv, hub = server + status, body = post(srv, "/api/mode", {"mode": MODE_MONITOR}) + assert status == 200 + assert json.loads(body)["mode"] == MODE_MONITOR + assert hub.mode == MODE_MONITOR + + +def test_invalid_mode_rejected(server): + srv, _ = server + with pytest.raises(urllib.error.HTTPError) as excinfo: + post(srv, "/api/mode", {"mode": "nonsense"}) + assert excinfo.value.code == 400 + + +def test_malformed_json_rejected(server): + srv, _ = server + request = urllib.request.Request( + base_url(srv) + "/api/mode", data=b"{not json", + method="POST", headers={"Content-Type": "application/json"}, + ) + with pytest.raises(urllib.error.HTTPError) as excinfo: + urllib.request.urlopen(request, timeout=5) + assert excinfo.value.code == 400 + + +# -- SSE -------------------------------------------------------------------- + +def test_sse_stream_delivers_a_tap(server): + """The stream used by the web page must push a tap as it happens.""" + srv, hub = server + received = [] + + def listen(): + with urllib.request.urlopen(base_url(srv) + "/api/events", timeout=10) as resp: + for raw in resp: + line = raw.decode("utf-8").strip() + if line.startswith("data: "): + received.append(json.loads(line[6:])) + return + + thread = threading.Thread(target=listen, daemon=True) + thread.start() + time.sleep(0.4) # let the subscription register + hub.publish(make_event()) + thread.join(timeout=6) + assert received and received[0]["id"] == 12886197 + + +def test_loopback_helper(): + assert bridge.is_loopback("127.0.0.1") is True + assert bridge.is_loopback("0.0.0.0") is False + + +# -- latched card (the "always available" behaviour) ------------------------ + +def test_card_not_present_before_any_tap(): + hub = EventHub() + state = hub.current_card() + assert state["present"] is False + assert state["card"] is None + + +def test_card_stays_available_after_the_page_loads(): + """The tap must outlive the page that reads it.""" + hub = EventHub() + hub.publish(make_event(value=555)) + for _ in range(3): # simulated later page loads + state = hub.current_card() + assert state["present"] is True + assert state["card"]["id"] == 555 + + +def test_card_cleared_when_removed(): + """Lifting the card must NOT lose the tap - only mark it as gone.""" + hub = EventHub(tap_ttl=30) + hub.publish(make_event()) + hub.publish(make_event("card_removed")) + state = hub.current_card() + assert state["present"] is False # no longer physically on the reader + assert state["claimable"] is True # but still spendable + assert state["card"]["id"] == 12886197 + + +def test_quick_tap_is_not_lost(): + """The real-world case: card lifted before the page polls.""" + hub = EventHub(tap_ttl=10) + hub.publish(make_event(value=31337)) + hub.publish(make_event("card_removed")) # lifted immediately + payload, error, _ = hub.consume_card() + assert error is None + assert payload["card"]["id"] == 31337 + + +def test_claim_returns_the_card_once(): + hub = EventHub() + hub.publish(make_event(value=777)) + payload, error, status = hub.consume_card() + assert error is None and status == 200 + assert payload["card"]["id"] == 777 + assert payload["consumed"] is True + + +def test_claim_is_single_use(): + """A page reload must not turn one physical tap into two logins.""" + hub = EventHub() + hub.publish(make_event()) + assert hub.consume_card()[1] is None + payload, error, status = hub.consume_card() + assert payload is None + assert error == "card already used" + assert status == 409 + + +def test_claim_without_a_card_is_conflict(): + hub = EventHub() + payload, error, status = hub.consume_card() + assert payload is None + assert error == "no card read yet" + assert status == 409 + + +def test_removing_and_replacing_the_card_allows_a_new_claim(): + """Lifting the card and tapping it again is a fresh authentication.""" + hub = EventHub(tap_ttl=30) + hub.publish(make_event(value=111)) + assert hub.consume_card()[1] is None + hub.publish(make_event("card_removed")) + hub.publish(make_event(value=111)) + payload, error, _ = hub.consume_card() + assert error is None + assert payload["card"]["id"] == 111 + + +def test_card_endpoint_reports_claimable_after_lift(server): + srv, hub = server + hub.publish(make_event(value=8080)) + hub.publish(make_event("card_removed")) + _, _, body = get(srv, "/api/card") + payload = json.loads(body) + assert payload["present"] is False + assert payload["claimable"] is True + assert payload["card"]["id"] == 8080 + + +def test_claim_expires_after_ttl(): + hub = EventHub(tap_ttl=0.05) + hub.publish(make_event()) + time.sleep(0.12) + payload, error, status = hub.consume_card() + assert payload is None + assert error == "card tap expired" + assert status == 410 + + +def test_ttl_zero_never_expires_while_present(): + hub = EventHub(tap_ttl=0) + hub.publish(make_event()) + state = hub.current_card() + assert state["expired"] is False + assert state["claimable"] is True + + +def test_tap_becomes_unclaimable_after_ttl(): + hub = EventHub(tap_ttl=0.05) + hub.publish(make_event()) + time.sleep(0.12) + assert hub.current_card()["claimable"] is False + + +def test_a_new_tap_replaces_an_unclaimed_one(): + hub = EventHub(tap_ttl=30) + hub.publish(make_event(value=111)) + hub.publish(make_event(value=222)) + payload, error, _ = hub.consume_card() + assert error is None + assert payload["card"]["id"] == 222 + + +def test_health_shows_card_presence(): + hub = EventHub() + hub.publish(make_event()) + health = hub.health() + assert health["card_present"] is True + assert health["card_consumed"] is False + + +# -- latched card over HTTP ------------------------------------------------- + +def test_card_endpoint_reports_presence(server): + srv, hub = server + _, _, body = get(srv, "/api/card") + assert json.loads(body)["present"] is False + + hub.publish(make_event(value=4242)) + _, _, body = get(srv, "/api/card") + payload = json.loads(body) + assert payload["present"] is True + assert payload["card"]["id"] == 4242 + + +def test_claim_endpoint_works_once(server): + srv, hub = server + hub.publish(make_event(value=9999)) + + status, body = post(srv, "/api/card/claim", {}) + assert status == 200 + assert json.loads(body)["card"]["id"] == 9999 + + with pytest.raises(urllib.error.HTTPError) as excinfo: + post(srv, "/api/card/claim", {}) + assert excinfo.value.code == 409 + detail = json.loads(excinfo.value.read()) + assert detail["error"] == "card already used" + + +def test_claim_endpoint_without_card_conflicts(server): + srv, _ = server + with pytest.raises(urllib.error.HTTPError) as excinfo: + post(srv, "/api/card/claim", {}) + assert excinfo.value.code == 409 + + +def test_claim_refused_from_other_origin(server): + """Another site must not be able to claim the card either.""" + srv, hub = server + hub.publish(make_event()) + with pytest.raises(urllib.error.HTTPError) as excinfo: + post(srv, "/api/card/claim", {}, + headers={"Origin": "https://evil.example"}) + assert excinfo.value.code == 403 + # the tap must still be claimable by the real caller + assert hub.current_card()["consumed"] is False + + +def test_claim_does_not_consume_on_refused_origin(server): + srv, hub = server + hub.publish(make_event(value=1234)) + with pytest.raises(urllib.error.HTTPError): + post(srv, "/api/card/claim", {}, + headers={"Origin": "https://evil.example"}) + payload, error, _ = hub.consume_card() + assert error is None + assert payload["card"]["id"] == 1234 diff --git a/tests/test_card.py b/tests/test_card.py new file mode 100644 index 0000000..2d5decb --- /dev/null +++ b/tests/test_card.py @@ -0,0 +1,111 @@ +"""The event model and its output formatting.""" + +import json + +import pytest + +from rdm_card_reader.card import ( + CONFIG_CARD_VALUE, + CardEvent, + format_event, +) + +SAMPLE = CardEvent( + event="card_inserted", + timestamp="2026-10-01 12:00:00", + value=0x00C4A0B5, + checksum=0xDE, + card_type=0x0F, + is_valid=True, + port="/dev/ttyS0", +) + + +def test_hex_value_is_zero_padded_to_eight_digits(): + assert SAMPLE.hex_value == "0x00C4A0B5" + + +def test_from_card_accepts_any_object_with_the_right_fields(): + class FakeCard: + value = 1234 + checksum = 5 + type = 15 + is_valid = True + + event = CardEvent.from_card("card_inserted", FakeCard(), "/dev/ttyUSB0") + assert event.value == 1234 + assert event.hex_value == "0x000004D2" + assert event.port == "/dev/ttyUSB0" + + +def test_to_dict_has_the_documented_keys(): + assert set(SAMPLE.to_dict()) == { + "event", "timestamp", "id", "hex", "type", "checksum", "is_valid", "port", + } + + +def test_to_json_is_parseable(): + assert json.loads(SAMPLE.to_json())["hex"] == "0x00C4A0B5" + + +def test_inserted_block_lists_the_card_data(): + text = format_event(SAMPLE) + assert "CARD INSERTED" in text + assert "12886197" in text + assert "0x00C4A0B5" in text + assert "0x0F" in text + assert "222" in text + assert "True" in text + + +def test_config_card_is_flagged_in_the_note(): + event = CardEvent( + event="card_inserted", + timestamp="2026-10-01 12:00:00", + value=CONFIG_CARD_VALUE, + checksum=0, + card_type=0x0F, + is_valid=True, + port="/dev/ttyS0", + ) + text = format_event(event) + assert "config card" in text + assert str(CONFIG_CARD_VALUE) in text + + +def test_ordinary_card_has_no_config_note(): + assert "config card" not in format_event(SAMPLE) + + +def test_removed_line_is_single_line(): + event = CardEvent( + event="card_removed", + timestamp="2026-10-01 12:00:00", + value=0x00C4A0B5, + checksum=0xDE, + card_type=0x0F, + is_valid=True, + port="/dev/ttyS0", + ) + text = format_event(event) + assert text.count("\n") == 0 + assert "CARD REMOVED" in text + + +def test_invalid_line_mentions_checksum(): + event = CardEvent( + event="invalid_card", + timestamp="2026-10-01 12:00:00", + value=0xDEADBEEF, + checksum=0, + card_type=0x0F, + is_valid=False, + port="/dev/ttyS0", + ) + assert "checksum mismatch" in format_event(event) + + +@pytest.mark.parametrize("value,expected", [(0, "0x00000000"), (0xFFFFFFFF, "0xFFFFFFFF")]) +def test_hex_value_boundaries(value, expected): + event = CardEvent("card_inserted", "-", value, 0, 0x0F, True, "p") + assert event.hex_value == expected diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..bb44737 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,93 @@ +"""CLI behaviour: exit codes, output modes and argument handling.""" + +import json + +import pytest + +from rdm_card_reader import cli + + +def test_list_ports_exits_zero(capsys): + assert cli.main(["--list-ports"]) == cli.EXIT_OK + out = capsys.readouterr().out + assert "/dev/ttyS0" in out + + +def test_demo_exits_zero(capsys): + assert cli.main(["--demo", "--demo-count", "1"]) == cli.EXIT_OK + assert "CARD INSERTED" in capsys.readouterr().out + + +def test_demo_json_emits_one_object_per_event(capsys): + cli.main(["--demo", "--demo-count", "1", "--json"]) + lines = [line for line in capsys.readouterr().out.splitlines() if line.strip()] + events = [json.loads(line) for line in lines if line.startswith("{")] + assert events + assert {"event", "timestamp", "id", "hex"} <= set(events[0]) + + +def test_demo_count_is_respected(capsys): + cli.main(["--demo", "--demo-count", "4"]) + assert capsys.readouterr().out.count("CARD INSERTED") >= 1 + + +def test_missing_library_exits_two(monkeypatch, capsys): + monkeypatch.setattr(cli, "available", lambda: False) + assert cli.main([]) == cli.EXIT_NO_LIBRARY + assert "rdm6300" in capsys.readouterr().err + + +def test_no_port_available_exits_one(monkeypatch, capsys): + monkeypatch.setattr(cli, "available", lambda: True) + monkeypatch.setattr(cli, "open_reader", lambda *a, **kw: None) + assert cli.main(["--port", "/dev/nope"]) == cli.EXIT_NO_PORT + assert "Could not open any serial port" in capsys.readouterr().err + + +def test_explicit_port_is_the_only_one_tried(monkeypatch): + monkeypatch.setattr(cli, "available", lambda: True) + tried = [] + + def fake_open(callback, ports=None, **kwargs): + tried.extend(ports or []) + return None + + monkeypatch.setattr(cli, "open_reader", fake_open) + cli.main(["--port", "/dev/ttyUSB0"]) + assert tried == ["/dev/ttyUSB0"] + + +def test_keyboard_interrupt_is_handled(monkeypatch, capsys): + monkeypatch.setattr(cli, "available", lambda: True) + + class Reader: + port = "/dev/ttyS0" + + def start(self): + raise KeyboardInterrupt + + def close(self): + self.closed = True + + reader = Reader() + monkeypatch.setattr(cli, "open_reader", lambda *a, **kw: reader) + + assert cli.main([]) == cli.EXIT_OK + assert reader.closed is True + assert "Stopped by user" in capsys.readouterr().out + + +def test_version_flag_exits_zero(capsys): + with pytest.raises(SystemExit) as exc: + cli.main(["--version"]) + assert exc.value.code == 0 + assert "rdm-card-reader" in capsys.readouterr().out + + +def test_json_mode_prints_json_from_the_callback(capsys): + printer = cli._make_printer(as_json=True) + from rdm_card_reader.card import CardEvent + + printer(CardEvent("card_inserted", "-", 1, 2, 15, True, "p")) + payload = json.loads(capsys.readouterr().out.strip()) + assert payload["id"] == 1 diff --git a/tests/test_e2e_bridge.py b/tests/test_e2e_bridge.py new file mode 100644 index 0000000..4ef3261 --- /dev/null +++ b/tests/test_e2e_bridge.py @@ -0,0 +1,177 @@ +"""End-to-end test: a real tap must be claimable by a web page at any time. + +Simulates the RDM6300 on a pseudo-terminal, starts the actual bridge the way +systemd would, then walks the exact flow a web page performs: + + card placed -> page loads -> polls /api/card -> claims it -> hand over once + +Also asserts the two things that matter for security: the tap survives until +claimed, and it cannot be claimed twice. +""" + +import json +import os +import pty +import signal +import struct +import subprocess +import sys +import threading +import time +import urllib.error +import urllib.request + +HERE = os.path.dirname(os.path.abspath(__file__)) +PROJECT = os.path.dirname(HERE) +VENV_PY = os.path.join(PROJECT, ".venv", "bin", "python") +PYTHON = VENV_PY if os.path.exists(VENV_PY) else sys.executable + +CARD_VALUE = 0x00C4A0B5 +EXPECTED_ID = 12886197 + + +def em4100_frame(value, card_type=0x0F): + """12 hex nibbles: type + 4 id bytes + XOR checksum of those 5 bytes.""" + id_bytes = struct.pack(">I", value) + checksum = card_type + for b in id_bytes: + checksum ^= b + frame = (bytes([card_type]) + id_bytes + bytes([checksum])).hex().upper() + assert len(frame) == 12, frame + return frame.encode("ascii") + + +def main(): + master, slave = pty.openpty() + serial_port = os.ttyname(slave) + + # Port 0 would be cleaner but the CLI needs a real number; pick a free one. + import socket + probe = socket.socket() + probe.bind(("127.0.0.1", 0)) + http_port = probe.getsockname()[1] + probe.close() + + proc = subprocess.Popen( + [PYTHON, "-m", "rdm_card_reader", "--serve", + "--port", serial_port, "--bridge-port", str(http_port), + "--mode", "login", "--tap-ttl", "15", + "--allow-origin", "https://kiosk.example"], + cwd=PROJECT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, bufsize=1, start_new_session=True, + ) + + output = [] + lock = threading.Lock() + + def pump(): + for line in proc.stdout: + with lock: + output.append(line.rstrip()) + + threading.Thread(target=pump, daemon=True).start() + + base = "http://127.0.0.1:{}".format(http_port) + results = {} + + def api(path, method="GET", headers=None): + request = urllib.request.Request( + base + path, method=method, + headers=headers or {}, + data=b"" if method == "POST" else None, + ) + with urllib.request.urlopen(request, timeout=5) as response: + body = response.read() + return response.status, (json.loads(body) if body else None) + + try: + # Wait for the HTTP server to come up. + deadline = time.time() + 15 + while time.time() < deadline: + try: + api("/api/health") + break + except Exception: + time.sleep(0.3) + + results["bridge started"] = True + + # The child needs a moment to open the serial port; frames written + # before that are discarded, which would look like a reader bug. + time.sleep(2.0) + + # --- no card yet ------------------------------------------------- + _, state = api("/api/card") + results["no card -> nothing claimable"] = state["claimable"] is False + + # --- a single tap, as a person would ------------------------------- + frame = em4100_frame(CARD_VALUE) + os.write(master, b"\x02" + frame + b"\x03") + time.sleep(0.3) + + # --- a page loads later and polls; the card must still be there ---- + time.sleep(1.0) + _, state = api("/api/card") + with lock: + print("state after tap: {}".format(json.dumps(state)), flush=True) + for line in output[-12:]: + print(" " + line) + results["card still available later"] = bool( + state["claimable"] and state["card"]["id"] == EXPECTED_ID + ) + results["card not consumed yet"] = state["consumed"] is False + + # --- claim it, as the web page does ------------------------------- + status, claimed = api( + "/api/card/claim", method="POST", + headers={"Origin": "https://kiosk.example"}, + ) + results["claim returns 200"] = status == 200 + results["claim gives the right card"] = claimed["card"]["id"] == EXPECTED_ID + + # --- single use --------------------------------------------------- + try: + api("/api/card/claim", method="POST", + headers={"Origin": "https://kiosk.example"}) + results["second claim refused"] = False + except urllib.error.HTTPError as exc: + results["second claim refused"] = exc.code == 409 + + # --- a foreign origin cannot grab it either ------------------------ + _, state = api("/api/card") + results["card marked consumed"] = state["consumed"] is True + + finally: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGINT) + except Exception: + pass + try: + proc.wait(timeout=8) + except Exception: + proc.kill() + try: + os.close(master) + os.close(slave) + except Exception: + pass + + with lock: + lines = list(output) + + print("======== bridge output ========") + for line in lines[:14]: + print(line) + print("===============================") + print() + + ok = True + for name, passed in results.items(): + print(" [{}] {}".format("PASS" if passed else "FAIL", name)) + ok = ok and bool(passed) + + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_e2e_pty.py b/tests/test_e2e_pty.py new file mode 100644 index 0000000..ac36ba1 --- /dev/null +++ b/tests/test_e2e_pty.py @@ -0,0 +1,165 @@ +"""End-to-end test over a pseudo-terminal, with no RFID hardware. + +The test owns the master side of a PTY, points the CLI at the slave side and +writes real RDM6300 frames. This exercises the same code path a physical +reader would: serial open -> byte stream -> parse -> callback -> stdout. +""" + +import json +import os +import pty +import signal +import subprocess +import sys +import threading +import time + +import pytest + +from rdm_card_reader.card import CONFIG_CARD_VALUE + +from .test_protocol import encode_stream, encode_frame # noqa: F401 + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +SRC = os.path.join(ROOT, "src") + +TIMEOUT = 15.0 + + +class ReaderProcess: + """Run the CLI against a PTY and collect its stdout in the background.""" + + def __init__(self, extra_args=()): + self.master, slave = pty.openpty() + self.port = os.ttyname(slave) + env = dict(os.environ) + env["PYTHONPATH"] = SRC + os.pathsep + env.get("PYTHONPATH", "") + self.proc = subprocess.Popen( + [sys.executable, "-u", "-m", "rdm_card_reader", + "--port", self.port, "--heartbeat", "0.3", *extra_args], + cwd=ROOT, env=env, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, text=True, bufsize=1, + start_new_session=True, + ) + self.lines = [] + self._lock = threading.Lock() + self._thread = threading.Thread(target=self._pump, daemon=True) + self._thread.start() + + def _pump(self): + for line in self.proc.stdout: + with self._lock: + self.lines.append(line.rstrip()) + + def output(self): + with self._lock: + return "\n".join(self.lines) + + def send(self, frame): + os.write(self.master, b"\x02" + frame + b"\x03") + + def wait_for(self, needle, timeout=TIMEOUT): + deadline = time.time() + timeout + while time.time() < deadline: + if needle in self.output(): + return True + time.sleep(0.05) + return False + + def __enter__(self): + self.wait_for("waiting for cards") + return self + + def __exit__(self, *exc): + try: + os.killpg(os.getpgid(self.proc.pid), signal.SIGINT) + except Exception: + pass + try: + self.proc.wait(timeout=5) + except Exception: + self.proc.kill() + self._thread.join(timeout=3) + for fd in (self.master, self.proc.stdout): + try: + fd.close() + except Exception: + pass + return False + + +@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") +def test_card_insert_and_remove_are_reported(): + card_id = 0x00C4A0B5 + frame = encode_frame(card_id) + + with ReaderProcess() as rp: + deadline = time.time() + TIMEOUT + while time.time() < deadline and not rp.wait_for("CARD INSERTED", 0.2): + rp.send(frame) + + assert rp.wait_for("CARD INSERTED"), rp.output() + assert rp.wait_for("CARD REMOVED") + + text = rp.output() + assert rp.port in text + assert str(card_id) in text + assert "0x00C4A0B5" in text + assert "222" in text # checksum = 0xDE + assert "0x0F" in text # type + assert "True" in text # valid + + +@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") +def test_corrupt_checksum_is_reported_as_invalid(): + good = bytearray(encode_frame(0x1A2B3C4D)) + good[-1] = ord("0") # break the checksum + bad = bytes(good) + + with ReaderProcess() as rp: + deadline = time.time() + TIMEOUT + while time.time() < deadline and not rp.wait_for("INVALID CARD", 0.2): + rp.send(bad) + + assert "INVALID CARD" in rp.output() + assert "checksum mismatch" in rp.output() + + +@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") +def test_json_mode_emits_machine_readable_events(): + with ReaderProcess(extra_args=("--json",)) as rp: + deadline = time.time() + TIMEOUT + while time.time() < deadline and not rp.wait_for('"event"', 0.2): + rp.send(encode_frame(0x00C4A0B5)) + + payloads = [ + json.loads(line) + for line in rp.output().splitlines() + if line.strip().startswith("{") + ] + assert payloads, rp.output() + assert any(p["event"] == "card_inserted" for p in payloads) + assert any(p["id"] == 0x00C4A0B5 for p in payloads) + assert all(p["port"] == rp.port for p in payloads) + + +@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") +def test_config_card_is_annotated(): + with ReaderProcess() as rp: + deadline = time.time() + TIMEOUT + while time.time() < deadline and not rp.wait_for("config card", 0.2): + rp.send(encode_frame(CONFIG_CARD_VALUE)) + + assert "config card" in rp.output() + + +@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") +def test_removal_after_heartbeat_expires(): + with ReaderProcess() as rp: + deadline = time.time() + TIMEOUT + while time.time() < deadline and not rp.wait_for("CARD INSERTED", 0.2): + rp.send(encode_frame(0x00C4A0B5)) + # stop sending and let the 0.3 s heartbeat lapse + assert rp.wait_for("CARD REMOVED", timeout=5) + + assert "CARD REMOVED" in rp.output() diff --git a/tests/test_ports.py b/tests/test_ports.py new file mode 100644 index 0000000..b1e8ede --- /dev/null +++ b/tests/test_ports.py @@ -0,0 +1,228 @@ +"""Tests for reliable serial-port identification. + +The point of these tests is that picking the wrong ttyUSB silently is the +failure mode that matters, so ambiguity must raise rather than guess. +""" + +import os +import sys + +SRC = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" +) +if SRC not in sys.path: + sys.path.insert(0, SRC) + +import pytest # noqa: E402 + +from rdm_card_reader import ports # noqa: E402 +from rdm_card_reader.ports import ( # noqa: E402 + AmbiguousPort, + PortInfo, + PortNotFound, + best_usb_candidates, + resolve, +) + + +def make_port(device, vid=None, pid=None, serial=None, description="", + aliases=None, manufacturer=None, product=None): + return PortInfo( + device=device, + vid=vid, + pid=pid, + serial_number=serial, + description=description, + aliases=list(aliases or []), + manufacturer=manufacturer, + product=product, + ) + + +@pytest.fixture +def two_adapters(monkeypatch): + """Two identical CH340 adapters plus an on-board UART.""" + a = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "CH340A", + "USB Serial", ["/dev/serial/by-id/usb-1a86_A-if00-port0"]) + b = make_port("/dev/ttyUSB1", 0x1A86, 0x7523, "CH340B", + "USB Serial", ["/dev/serial/by-id/usb-1a86_B-if00-port0"]) + onboard = make_port("/dev/ttyS0", description="ttyS0") + monkeypatch.setattr(ports, "discover", lambda: [a, b, onboard]) + return a, b, onboard + + +# -- identity --------------------------------------------------------------- + +def test_usb_id_and_chip_name(): + port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523) + assert port.usb_id == "1a86:7523" + assert port.chip == "CH340" + assert port.is_usb is True + + +def test_unknown_chip_still_identified_as_usb(): + port = make_port("/dev/ttyUSB0", 0xFFFF, 0x0001) + assert port.chip is None + assert port.is_usb is True + + +def test_stable_id_prefers_by_id_symlink(): + port = make_port( + "/dev/ttyUSB0", + aliases=["/dev/serial/by-path/pci-1-usb-0:1.2:1.0-port0", + "/dev/serial/by-id/usb-1a86_USB-if00-port0"], + ) + assert port.stable_id == "/dev/serial/by-id/usb-1a86_USB-if00-port0" + assert port.by_path.endswith("port0") + + +def test_onboard_uart_is_not_usb(): + assert make_port("/dev/ttyS0").is_usb is False + + +# -- scoring ---------------------------------------------------------------- + +def test_usb_adapter_scores_above_onboard_uart(): + adapter = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "SN1") + onboard = make_port("/dev/ttyS0") + assert adapter.score() > onboard.score() + + +def test_serial_number_raises_score(): + with_sn = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "SN1") + without = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, None) + assert with_sn.score() > without.score() + + +# -- matching --------------------------------------------------------------- + +def test_match_by_device_path(): + assert make_port("/dev/ttyUSB0").matches("/dev/ttyUSB0") + + +def test_match_by_serial_number(): + port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "CH340A") + assert port.matches("CH340A") + + +def test_match_by_vid_pid(): + port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523) + assert port.matches("1a86:7523") + + +def test_match_by_chip_name(): + assert make_port("/dev/ttyUSB0", 0x1A86, 0x7523).matches("ch340") + + +def test_match_by_id_symlink(): + port = make_port("/dev/ttyUSB0", + aliases=["/dev/serial/by-id/usb-1a86_A-if00-port0"]) + assert port.matches("/dev/serial/by-id/usb-1a86_A-if00-port0") + + +def test_empty_selector_never_matches(): + assert not make_port("/dev/ttyUSB0").matches("") + + +# -- resolution ------------------------------------------------------------- + +def test_resolve_by_serial_number_picks_right_adapter(two_adapters): + device, reason = resolve(selectors=["CH340B"]) + assert device == "/dev/ttyUSB1" + assert "matched" in reason + + +def test_resolve_by_by_id_path(two_adapters): + device, _ = resolve(selectors=["/dev/serial/by-id/usb-1a86_B-if00-port0"]) + assert device == "/dev/ttyUSB1" + + +def test_resolve_refuses_when_selector_matches_both(two_adapters): + """A vid:pid shared by two adapters must not silently pick one.""" + with pytest.raises(AmbiguousPort): + resolve(selectors=["1a86:7523"]) + + +def test_resolve_unknown_selector_raises_not_found(two_adapters): + with pytest.raises(PortNotFound): + resolve(selectors=["NOSUCHDEVICE"]) + + +def test_resolve_falls_back_to_raw_path_when_not_enumerable(monkeypatch, tmp_path): + fake = tmp_path / "myport" + fake.write_text("") + monkeypatch.setattr(ports, "discover", lambda: []) + device, reason = resolve(selectors=[str(fake)]) + assert device == str(fake) + assert "explicit path" in reason + + +def test_autodetect_requires_exactly_one(monkeypatch): + single = make_port("/dev/ttyUSB3", 0x1A86, 0x7523, "ONLY") + monkeypatch.setattr(ports, "discover", lambda: [single, make_port("/dev/ttyS0")]) + device, reason = resolve(autodetect=True) + assert device == "/dev/ttyUSB3" + assert "autodetected" in reason + + +def test_autodetect_refuses_with_two_adapters(two_adapters): + with pytest.raises(AmbiguousPort): + resolve(autodetect=True) + + +def test_autodetect_ignores_onboard_uart(monkeypatch): + """An on-board UART must not make a single USB adapter look ambiguous.""" + adapter = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "ONLY") + monkeypatch.setattr( + ports, "discover", + lambda: [adapter, make_port("/dev/ttyS0"), make_port("/dev/ttyAMA0")], + ) + device, _ = resolve(autodetect=True) + assert device == "/dev/ttyUSB0" + + +def test_fallback_order_used_when_no_usb(monkeypatch): + """Must not depend on the host actually having a serial port.""" + monkeypatch.setattr(ports, "discover", lambda: []) + monkeypatch.setattr(ports.os.path, "exists", lambda path: path == "/dev/ttyS0") + device, reason = resolve(fallback_ports=["/dev/ttyS0", "/dev/ttyUSB0"]) + assert device == "/dev/ttyS0" + assert "default order" in reason + + +def test_fallback_order_skips_missing_ports(monkeypatch): + monkeypatch.setattr(ports, "discover", lambda: []) + monkeypatch.setattr( + ports.os.path, "exists", lambda path: path == "/dev/ttyUSB0" + ) + device, _ = resolve(fallback_ports=["/dev/ttyS0", "/dev/ttyUSB0"]) + assert device == "/dev/ttyUSB0" + + +def test_no_port_at_all_raises(monkeypatch): + monkeypatch.setattr(ports, "discover", lambda: []) + with pytest.raises(PortNotFound): + resolve(fallback_ports=[]) + + +# -- helper ---------------------------------------------------------------- + +def test_best_usb_candidates_dedupes_same_identity(): + a = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "SAME") + dup = make_port("/dev/ttyUSB1", 0x1A86, 0x7523, "SAME") + result = best_usb_candidates([a, dup]) + assert len(result) == 1 + + +def test_best_usb_candidates_keeps_distinct_serial_numbers(): + a = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "A") + b = make_port("/dev/ttyUSB1", 0x1A86, 0x7523, "B") + assert len(best_usb_candidates([a, b])) == 2 + + +def test_describe_is_human_readable(): + port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "ABC", "USB Serial") + text = port.describe() + assert "/dev/ttyUSB0" in text + assert "CH340" in text + assert "ABC" in text diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..927cc7a --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,108 @@ +"""RDM6300 wire format: type + id + checksum, 12 hex nibbles per frame. + +The checksum is the XOR of the type byte *and* the four id bytes. Getting this +wrong (e.g. XOR-ing only the id) makes every card read fail its checksum, so +the exact bytes are pinned here. +""" + +import struct + +import pytest + +from rdm_card_reader.card import CardData + +START = 0x02 +END = 0x03 + + +def encode_frame(value, card_type=0x0F): + """Encode a card as the 12 nibbles the RDM6300 emits between STX and ETX.""" + id_bytes = struct.pack(">I", value) + checksum = card_type + for byte in id_bytes: + checksum ^= byte + frame = bytes([card_type]) + id_bytes + bytes([checksum]) + encoded = frame.hex().upper().encode("ascii") + assert len(encoded) == 12 + return encoded + + +def encode_stream(value, card_type=0x0F): + return bytes([START]) + encode_frame(value, card_type) + bytes([END]) + + +def decode_frame(frame): + """Reference decoder mirroring rdm6300.BaseReader._parse_fragment.""" + nibbles = [int(ch, 16) for ch in frame.decode("ascii")] + + def to_int(chunk): + out = 0 + for item in chunk: + out = (out << 4) | item + return out + + calculated = 0 + for i in range(0, 10, 2): + calculated ^= (nibbles[i] << 4) | nibbles[i + 1] + received = to_int(nibbles[10:12]) + + return CardData( + value=to_int(nibbles[2:10]), + checksum=received, + is_valid=received == calculated, + type=to_int(nibbles[0:2]), + ) + + +@pytest.mark.parametrize( + "value,expected", + [ + (0x00C4A0B5, b"0F00C4A0B5DE"), + (0x1A2B3C4D, b"0F1A2B3C4D4F"), + ], +) +def test_encoded_frame_bytes_are_stable(value, expected): + assert encode_frame(value) == expected + + +def test_frame_is_twelve_nibbles(): + assert len(encode_frame(0x00C4A0B5)) == 12 + + +def test_checksum_covers_type_byte_not_only_id(): + """Same id with a different type byte must change the checksum.""" + assert encode_frame(0x00C4A0B5, 0x0F) != encode_frame(0x00C4A0B5, 0x10) + + +def test_round_trip_yields_valid_card(): + card = decode_frame(encode_frame(0x00C4A0B5)) + assert card.value == 0x00C4A0B5 + assert card.type == 0x0F + assert card.checksum == 0xDE + assert card.is_valid is True + + +def test_round_trip_of_second_sample(): + card = decode_frame(encode_frame(0x1A2B3C4D)) + assert card.value == 0x1A2B3C4D + assert card.is_valid is True + + +def test_corrupted_checksum_is_invalid(): + frame = bytearray(encode_frame(0x00C4A0B5)) + frame[-1] = ord("0") + assert decode_frame(bytes(frame)).is_valid is False + + +def test_frame_with_wrong_nibble_count_is_rejected(): + """A 14 nibble frame leaves 2 nibbles buffered in the real reader.""" + assert len(encode_frame(0x00C4A0B5)) == 12 + fourteen = b"0F00C4A0B5DE00" + assert len(fourteen) == 14 + + +def test_stream_wrapper_uses_stx_and_etx(): + stream = encode_stream(0x00C4A0B5) + assert stream[0] == START + assert stream[-1] == END + assert stream[1:-1] == encode_frame(0x00C4A0B5) diff --git a/tests/test_reader.py b/tests/test_reader.py new file mode 100644 index 0000000..398fca0 --- /dev/null +++ b/tests/test_reader.py @@ -0,0 +1,149 @@ +"""Port selection and error explanation, using a fake reader library.""" + +import types + +import pytest + +import rdm_card_reader.reader as reader_module +from rdm_card_reader.card import CANDIDATE_PORTS +from rdm_card_reader.reader import explain_error, list_ports, open_reader + + +class FakeBaseReader: + """Stands in for rdm6300.BaseReader; records the port it was built with.""" + + fail_on = () + + def __init__(self, port, heartbeat_interval=0.5): + if port in type(self).fail_on: + raise OSError("could not open port {}".format(port)) + self.port = port + self.heartbeat_interval = heartbeat_interval + self.started = False + self.closed = False + + def start(self): + self.started = True + + def close(self): + self.closed = True + + +@pytest.fixture +def fake_library(monkeypatch): + module = types.SimpleNamespace(BaseReader=FakeBaseReader) + monkeypatch.setattr(reader_module, "rdm6300", module) + return module + + +@pytest.fixture(autouse=True) +def restore_library(): + original = reader_module.rdm6300 + yield + reader_module.rdm6300 = original + + +def test_default_port_order_matches_the_workstation_app(): + assert CANDIDATE_PORTS == ( + "/dev/ttyS0", "/dev/ttyAMA0", "/dev/ttyUSB0", "/dev/ttyACM0", + ) + + +def test_first_working_port_wins(fake_library): + FakeBaseReader.fail_on = ("/dev/ttyS0", "/dev/ttyAMA0") + reader = open_reader(lambda event: None, + ports=["/dev/ttyS0", "/dev/ttyAMA0", "/dev/ttyUSB0"]) + assert reader is not None + assert reader.port == "/dev/ttyUSB0" + + +def test_returns_none_when_no_port_works(fake_library): + FakeBaseReader.fail_on = ("/dev/ttyS0", "/dev/ttyAMA0") + assert open_reader(lambda event: None, + ports=["/dev/ttyS0", "/dev/ttyAMA0"]) is None + + +def test_failures_are_reported_with_the_port(fake_library): + FakeBaseReader.fail_on = ("/dev/ttyS0",) + seen = [] + open_reader(lambda event: None, ports=["/dev/ttyS0"], + on_error=lambda port, message: seen.append((port, message))) + assert seen and seen[0][0] == "/dev/ttyS0" + + +def test_heartbeat_interval_is_passed_through(fake_library): + FakeBaseReader.fail_on = () + reader = open_reader(lambda event: None, ports=["/dev/ttyS0"], + heartbeat_interval=1.5) + assert reader.heartbeat_interval == 1.5 + + +def test_heartbeat_none_disables_removal_detection(fake_library): + FakeBaseReader.fail_on = () + reader = open_reader(lambda event: None, ports=["/dev/ttyS0"], + heartbeat_interval=None) + assert reader.heartbeat_interval is None + + +def test_events_are_forwarded_to_the_callback(fake_library): + FakeBaseReader.fail_on = () + received = [] + reader = open_reader(received.append, ports=["/dev/ttyS0"]) + + class Card: + value = 7 + checksum = 1 + type = 15 + is_valid = True + + reader.card_inserted(Card()) + assert len(received) == 1 + assert received[0].value == 7 + assert received[0].event == "card_inserted" + + +def test_card_removed_with_none_is_ignored(fake_library): + FakeBaseReader.fail_on = () + received = [] + reader = open_reader(received.append, ports=["/dev/ttyS0"]) + reader.card_removed(None) + assert received == [] + + +def test_invalid_card_is_forwarded(fake_library): + FakeBaseReader.fail_on = () + received = [] + reader = open_reader(received.append, ports=["/dev/ttyS0"]) + + class Card: + value = 1 + checksum = 2 + type = 15 + is_valid = False + + reader.invalid_card(Card()) + assert received[0].event == "invalid_card" + + +def test_missing_library_raises(monkeypatch): + monkeypatch.setattr(reader_module, "rdm6300", None) + with pytest.raises(reader_module.ReaderUnavailable): + open_reader(lambda event: None) + + +@pytest.mark.parametrize( + "message,expected_fragment", + [ + ("[Errno 13] Permission denied: '/dev/ttyS0'", "dialout"), + ("could not open port /dev/nope", "device not found"), + ("something else", "something else"), + ], +) +def test_explain_error(message, expected_fragment): + assert expected_fragment in explain_error("/dev/ttyS0", message) + + +def test_list_ports_shape(): + info = list_ports() + assert set(info) == {"candidates", "detected"} + assert [port for port, _ in info["candidates"]] == list(CANDIDATE_PORTS)