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.
This commit is contained in:
2026-10-01 15:31:04 +03:00
commit 576f22268f
26 changed files with 3771 additions and 0 deletions
+12
View File
@@ -0,0 +1,12 @@
__pycache__/
*.py[cod]
*.egg-info/
build/
dist/
.venv/
venv/
.pytest_cache/
.coverage
htmlcov/
.idea/
.vscode/
+21
View File
@@ -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.
+376
View File
@@ -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 <card-ids-are-not-secrets>`: 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.
<a name="card-ids-are-not-secrets"></a>
> **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
+28
View File
@@ -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"
+162
View File
@@ -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 <<EOF
$(log "Installed")
Bridge : http://127.0.0.1:$BRIDGE_PORT
Card state : curl http://127.0.0.1:$BRIDGE_PORT/api/card
Health : curl http://127.0.0.1:$BRIDGE_PORT/api/health
Logs : journalctl -u rdm-card-reader -f
Test a tap : place a card and run the curl above
The service starts automatically on boot. The card stays readable at
/api/card while it is on the reader; POST /api/card/claim hands it over
once, so a page reload cannot replay the same tap into a second login.
EOF
+53
View File
@@ -0,0 +1,53 @@
[Unit]
Description=RDM card reader HTTP bridge (serves card taps for web login)
Documentation=file:/opt/rdm-card-reader/README.md
# The serial device must exist before we start, otherwise the reader opens the
# wrong tty or none at all.
After=dev-serial.device network-online.target
Wants=network-online.target
# Do not restart forever if the reader is genuinely absent. These are [Unit]
# options, not [Service] - systemd ignores them in the wrong section.
StartLimitIntervalSec=60
StartLimitBurst=5
[Service]
Type=simple
User=rdmcard
Group=dialout
WorkingDirectory=/opt/rdm-card-reader
# --serve exposes the card on http://127.0.0.1:8765/api/card
# Autodetect refuses to guess if more than one USB adapter is present, which is
# the safe behaviour: a wrong guess means authenticating the wrong person.
ExecStart=/opt/rdm-card-reader/.venv/bin/rdm-card-reader \
--serve \
--port /dev/rdm-reader \
--mode login \
--allow-origin https://your.web.app
Restart=always
RestartSec=3
# The bridge only needs to read the serial port and listen on loopback.
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
RestrictNamespaces=true
LockPersonality=true
MemoryDenyWriteExecute=true
DeviceAllow=/dev/rdm-reader rw
DevicePolicy=closed
# stdout/stderr go here; watch with: journalctl -u rdm-card-reader -f
StandardOutput=journal
StandardError=journal
SyslogIdentifier=rdm-card-reader
[Install]
WantedBy=multi-user.target
+40
View File
@@ -0,0 +1,40 @@
[build-system]
requires = ["setuptools>=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"
+51
View File
@@ -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__",
]
+8
View File
@@ -0,0 +1,8 @@
"""Allow ``python -m rdm_card_reader``."""
import sys
from .cli import main
if __name__ == "__main__":
sys.exit(main())
+503
View File
@@ -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")
+135
View File
@@ -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
)
+285
View File
@@ -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())
+49
View File
@@ -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
+397
View File
@@ -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"
)
)
View File
+111
View File
@@ -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}
View File
+8
View File
@@ -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)
+501
View File
@@ -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
+111
View File
@@ -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
+93
View File
@@ -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
+177
View File
@@ -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())
+165
View File
@@ -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()
+228
View File
@@ -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
+108
View File
@@ -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)
+149
View File
@@ -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)