openvibers/openvibe.link Commits

Node: the computer plugin — the machine's terminal, files and screen for its owner's agents, at the level the owner sets on the machine (#41)

OP OpenVibers committed 30 days ago · dc7d309160 · 24 files, +1599 −39

* Node: the computer plugin — the machine's terminal, files and screen for its owner's agents, at the level the owner sets on the machine

Plan T14/T17: the Node's computer control. This adds a `computer` plugin that gives the machine's terminal, files and screen to its owner's agents, under a control level the owner sets **on the machine**. Nothing in the cloud can raise that level.

**Control levels**
- `observe` (the default): screen; list, stat and read inside the folders.
- `ask`: also run programs and write files, each only after `openvibe-node approve <id>` on the machine.
- `trusted`: runs and writes without asking, inside the folders.
- `full`: anywhere the account can reach.

**What the plugin enforces**
- It never runs as root.
- Paths are checked after symlinks are resolved.
- Programs are argv only (never a shell line), run in their own process group, at most 2 at a time, and are killed at the owner's timeout. Output is capped.
- The local kill switch or an e-stop drops it to `observe`: what runs is killed and what waits is denied.

**Core**
- Kinds `exec`, `fs`, `screen` (capabilities `terminal`, `files`, `screen`).
  - Long reply windows: 60 s by default, capped at 10 min. No clamping; the value is a JSON object of at most 384 KB.
  - `exec` and file writes are refused in the core while a stop is latched.
  - Acks carry the answer in `data` (`protocol.AckData`, at most 512 KB). An ack without data still decodes as `Ack`.
- Approvals: the plugin's `approval` / `approval_done` events are tracked. New local-control commands `approvals`, `approve`, `deny` forward the answer to the plugin.
- CLI: `openvibe-node approvals [--json]`, `approve <ID>`, `deny <ID>`.

**SDK**
- `handle()` may return a dict (the ack's `data`), or raise `Deferred` and answer later with `complete()` / `fail()` from any thread.
- New `on_approval(id, approved)`.
- Kinds that move nothing keep their full deadline; motion kinds stay capped at 5 s.

**Installer**
- `--computer observe|ask|trusted|full` (Linux): the service runs as the installing person's own account, with their home as the only folder.
- Refused as root and together with `--worker`.
- `scripts/test-install-computer-config.sh` is added to CI.

**Docs:** new `docs/computer.md`; updates to `docs/protocol.md`, `docs/plugins.md` and the README.

**Tests**
- 13 plugin tests: levels, roots and symlinks, approve/deny/expiry, kill switch, halt, caps and timeouts, JPEG screen.
- 2 new SDK tests: data, `Deferred`, approval op, deadline caps.
- 4 end-to-end Go tests with the real plugin through the core and the fake Bot: trusted exec with data, ask → approve/deny over the control socket, local stop leaves observe only, no plugin → unsupported.
- `go test -race ./...` passes; so do shellcheck and the installer tests.

Bot routing these kinds for a person's machines comes next, in Contracts + Bot.

* Node: the computer plugin reports its level and what waits as sensors (the core forwards sensors to Bot)

* Node: the computer end-to-end tests are unix-only, like the other end-to-end tests

Changes

ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex acd4e10..3fb28b3 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -91,18 +91,20 @@ jobs:
- run: python -m pytest -q plugins/cozmo/tests
- run: python -m pytest -q plugins/relay/tests
- run: python -m pytest -q plugins/ros2/tests
++ - run: python -m pytest -q plugins/computer/tests
- name: packages install
-- run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay ./plugins/ros2
++ run: pip install ./plugins/sdk ./plugins/dryrun ./plugins/adeept_adr036 ./plugins/cozmo ./plugins/relay ./plugins/ros2 ./plugins/computer
shell:
name: shellcheck
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
-- - run: shellcheck -s sh install/install.sh scripts/dist.sh scripts/test-install-stock-server.sh scripts/test-install-pair-args.sh scripts/test-install-worker-config.sh
++ - run: shellcheck -s sh install/install.sh scripts/dist.sh scripts/test-install-stock-server.sh scripts/test-install-pair-args.sh scripts/test-install-worker-config.sh scripts/test-install-computer-config.sh
- run: sh scripts/test-install-stock-server.sh
- run: sh scripts/test-install-pair-args.sh
- run: sh scripts/test-install-worker-config.sh
++ - run: sh scripts/test-install-computer-config.sh
esp32-conformance:
name: esp32 protocol conformance (desktop)
ddiff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
iindex 8d1a8b4..c31d98e 100644
---- a/.github/workflows/release.yml
++++ b/.github/workflows/release.yml
@@@ -23,7 +23,7 @@ jobs:
- run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 || true
- run: go vet ./... && go test -count=1 ./...
- run: |
-- for p in sdk dryrun adeept_adr036 cozmo relay ros2; do python -m pytest -q plugins/$p/tests; done
++ for p in sdk dryrun adeept_adr036 cozmo relay ros2 computer; do python -m pytest -q plugins/$p/tests; done
- run: scripts/dist.sh "${GITHUB_REF_NAME}"
# Every release must carry the installer OpenVibe.Bot's GET /install route and BOT_INSTALLER_SOURCE_URL are
# documented to serve (releases/latest/download/install.sh), the plugin bundle the installer downloads, and the
ddiff --git a/README.md b/README.md
iindex 9c55e6c..30f865d 100644
---- a/README.md
++++ b/README.md
@@@ -1,8 +1,8 @@
# OpenVibe Node
One service for your computers, servers, Raspberry Pis, phones and robots: pair it once with OpenVibe and products
--like **OpenVibe.Bot** (robots: drive, cameras, sensors, with safety) and, later, **OpenVibe.Actor** (computer control
--for AI agents) can use the device. It connects out, needs no open ports, and keeps a local kill switch.
++like **OpenVibe.Bot** (robots: drive, cameras, sensors, with safety) and **OpenVibe.Actor** (computer control for AI
++agents, at the level you set on the machine) can use the device. It connects out, needs no open ports, and keeps a local kill switch.
```sh
curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept
@@@ -17,6 +17,7 @@ install, Windows and configuration: [docs/install.md](docs/install.md).
openvibe-node status # link, plugins, video, battery, stop latch
openvibe-node stop # kill switch: stop every motor now and keep it stopped
openvibe-node resume
++openvibe-node approvals # the computer plugin at level ask: what waits; approve <id> / deny <id>
```
## What is in here
@@@ -40,6 +41,10 @@ signature against the key Bot publishes (`--key-id` pins one) and names the plug
on a community profile runs the same check and warns when it fails. Install a community plugin only when it passes
([docs/plugins.md](docs/plugins.md#community-plugins)).
++**Your computer for your agents:** `--computer observe|ask|trusted|full` adds the `computer` plugin (terminal, files
++and screen for OpenVibe.Actor), running as your own account, at the level you set on the machine; `ask` waits for
++`openvibe-node approve <id>` ([docs/computer.md](docs/computer.md)).
++
**A machine that runs your jobs:** add it on openvibe.bot/machines, which prints the same installer with `--worker`: the
job worker runs each job as an unprivileged user in a sandbox, and the OpenVibe.Codes job runner lets you start a coding
task on openvibe.codes/run from any browser and have this machine do it ([docs/worker.md](docs/worker.md)).
@@@ -53,7 +58,7 @@ Decisions: [ADR-043](docs/ADR-043-bot-devices-and-control.md).
```sh
go vet ./... && go test -race ./... # the end-to-end tests run the Python dry-run plugin
pip install pytest pillow numpy
--for p in sdk dryrun adeept_adr036 cozmo relay; do python3 -m pytest -q plugins/$p/tests; done
++for p in sdk dryrun adeept_adr036 cozmo relay ros2 computer; do python3 -m pytest -q plugins/$p/tests; done
scripts/dist.sh v0.0.0-dev # every binary, the plugin bundle and SHA256SUMS in dist/
```
ddiff --git a/cmd/openvibe-node/main.go b/cmd/openvibe-node/main.go
iindex 0ac9ba0..0fb6b73 100644
---- a/cmd/openvibe-node/main.go
++++ b/cmd/openvibe-node/main.go
@@@ -57,6 +57,9 @@ Usage:
openvibe-node stop stop every actuator now and hold them stopped
openvibe-node resume release the stop
openvibe-node plugins [--json]
++ openvibe-node approvals [--json] what the computer plugin waits to be approved (control level "ask")
++ openvibe-node approve <ID> let it run
++ openvibe-node deny <ID> refuse it
openvibe-node profile verify community.<name> [--server URL] [--key-id ID]
check a community robot profile against the signature Bot made when it was reviewed
openvibe-node version
@@@ -178,6 +181,14 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
err = cmdResume(g, stdout)
case "plugins":
err = cmdPlugins(g, *asJSON, stdout)
++ case "approvals":
++ err = cmdApprovals(g, *asJSON, stdout)
++ case "approve", "deny":
++ if len(positional) != 1 {
++ fmt.Fprintf(stderr, "usage: openvibe-node %s <ID> (openvibe-node approvals lists the ids)\n", cmd)
++ return 2
++ }
++ err = cmdAnswer(g, cmd, positional[0], stdout)
case "profile":
if len(positional) != 2 || positional[0] != "verify" {
fmt.Fprintln(stderr, "usage: openvibe-node profile verify community.<name> [--server URL] [--key-id ID]")
@@@ -596,6 +607,56 @@ func cmdResume(g *globals, stdout io.Writer) error {
return nil
}
++// cmdApprovals lists what the computer plugin holds for its owner's answer (control level "ask").
++func cmdApprovals(g *globals, asJSON bool, stdout io.Writer) error {
++ resp, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdApprovals}, 2*time.Second)
++ if err != nil {
++ return fmt.Errorf("the node is not running (%v)", err)
++ }
++ if !resp.OK {
++ return errors.New(resp.Error)
++ }
++ if asJSON {
++ _, err := stdout.Write(append(resp.Data, '\n'))
++ return err
++ }
++ var list []node.Approval
++ if err := json.Unmarshal(resp.Data, &list); err != nil {
++ return err
++ }
++ if len(list) == 0 {
++ fmt.Fprintln(stdout, "Nothing waits for approval.")
++ return nil
++ }
++ for _, a := range list {
++ age := time.Since(time.UnixMilli(a.At)).Round(time.Second)
++ fmt.Fprintf(stdout, "%s %s, %s ago: %s\n", a.ID, a.Kind, age, a.Summary)
++ }
++ fmt.Fprintln(stdout, "\nopenvibe-node approve <ID> lets one run; openvibe-node deny <ID> refuses it.")
++ return nil
++}
++
++// cmdAnswer approves or denies one waiting computer command.
++func cmdAnswer(g *globals, cmd, id string, stdout io.Writer) error {
++ c := localctl.CmdApprove
++ if cmd == "deny" {
++ c = localctl.CmdDeny
++ }
++ resp, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: c, ID: strings.TrimSpace(id)}, 2*time.Second)
++ if err != nil {
++ return fmt.Errorf("the node is not running (%v)", err)
++ }
++ if !resp.OK {
++ return errors.New(resp.Error)
++ }
++ if c == localctl.CmdApprove {
++ fmt.Fprintln(stdout, "Approved: it runs now.")
++ } else {
++ fmt.Fprintln(stdout, "Denied: it does not run.")
++ }
++ return nil
++}
++
func cmdStatus(g *globals, asJSON bool, stdout io.Writer) error {
resp, err := localctl.Call(g.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStatus}, 2*time.Second)
if err != nil {
ddiff --git a/docs/computer.md b/docs/computer.md
nnew file mode 100644
iindex 0000000..49c849e
---- /dev/null
++++ b/docs/computer.md
@@@ -0,0 +1,81 @@
++# The computer plugin: your machine for your agents
++
++The `computer` plugin gives the machine's terminal, files and screen to its owner's agents (OpenVibe.Actor, plan
++T14/T17), through OpenVibe.Bot and the Node's one outbound link. It is **off** unless you turn it on, it never runs
++as root, and **you set what it may do on the machine itself**: nothing in the cloud can raise that level.
++
++## Turn it on
++
++On Linux, from your own account (the installer uses sudo itself):
++
++```sh
++curl -fsSL https://openvibe.bot/install | sh -s -- --network <URL> --pairing pair_… --code ABCD-1234 --computer ask
++```
++
++`--computer LEVEL` runs the Node's service as **your** account (never root, never the `openvibe-node` system account)
++and adds the plugin with your home folder as its only folder. It cannot share a machine with `--worker` yet (the job
++worker's service runs as root).
++
++By hand, or on another system, add it to `plugins` in the Node's `config.json` and run the Node as your own account
++(`openvibe-node run` in your session, or `openvibe-node install --user <you>`):
++
++```json
++{"name": "computer", "config": {"level": "ask", "roots": ["/home/you/projects"]}}
++```
++
++## Control levels
++
++| level | screen, list, stat, read (inside the folders) | run programs, write files (inside the folders) | outside the folders |
++|-----------|----------------------------------------------|-----------------------------------------------|---------------------|
++| `observe` | yes | no (`nack not_allowed`) | no |
++| `ask` | yes | after you approve each one on the machine | no |
++| `trusted` | yes | yes | no |
++| `full` | yes | yes | yes |
++
++`observe` is the default. Folders are `roots` (default: the home folder of the account running the Node); a path is
++checked after symlinks are resolved, so a link out of a folder is outside it.
++
++**Approvals** (level `ask`): a request waits on the machine until you answer it, for at most `approval_timeout_s`
++(default 120 s) and never longer than the request's own deadline:
++
++```sh
++openvibe-node approvals # what waits: id, kind, what it would do
++openvibe-node approve <ID> # let it run
++openvibe-node deny <ID> # refuse it
++```
++
++At most 8 requests wait at once. The panel on openvibe.bot sees `approval` and `approval_done` events, so it can say
++that a request is waiting on the machine.
++
++**Kill switch:** `openvibe-node stop` (and an owner's e-stop) drops the plugin to `observe`: what runs is killed, what
++waits is denied, and the core also refuses `exec` and file writes until `openvibe-node resume`. `halt` kills what runs.
++
++## What it does
++
++| kind | value | answer (`ack.data`) |
++|----------|-----------------------------------------------------------------------------------------------|---------------------|
++| `exec` | `argv` (a program and its arguments: never a shell line), `cwd`, `timeout_s`, `stdin` (≤ 64 KB) | `exit_code`, `stdout`, `stderr` (each cut at `max_output_bytes`), `truncated`, `timed_out`, `duration_ms` |
++| `fs` | `op` `list` / `stat` / `read` (`offset`, `length`) / `write` (`content_b64`, `mode` `create`, `overwrite` or `append`) / `mkdir`, and an absolute `path` | `entries` (≤ 1000), the stat, `content_b64` (≤ `max_read_bytes`), `written`, `created` |
++| `screen` | `max_width` (default 1280) | `format` (`jpeg` with Pillow, else `png` ≤ 300 KB), `width`, `height`, `content_b64` |
++
++A program runs in its own process group with the Node account's rights, at most 2 at a time, and is killed at
++`timeout_s` (the owner's `timeout_s` is the cap). A write with `overwrite` replaces the file atomically. Screens come
++from `screencapture` (macOS), `grim` (Wayland) or `import`, `scrot` or `gnome-screenshot` (X11), so the Node must run in
++your graphical session for `screen`; without one the plugin simply does not offer it.
++
++## Config
++
++| key | default | meaning |
++|----------------------|----------------|---------|
++| `level` | `observe` | `observe`, `ask`, `trusted` or `full` |
++| `roots` | your home | the folders it may use (absolute) |
++| `allow_exec` | `true` | `false`: never run programs (no `terminal` capability) |
++| `timeout_s` | 30 (≤ 600) | the longest a program runs |
++| `max_output_bytes` | 65536 (≤ 192 KB) | stdout and stderr are each cut there |
++| `max_read_bytes` | 262144 (≤ 256 KB) | the most one read or write moves |
++| `approval_timeout_s` | 120 (≤ 600) | how long a request waits for you |
++
++`describe` reports `terminal`, `files` and `screen` with the level and folders, so Bot shows what the machine allows;
++telemetry carries the sensors `computer_level` (observe while a stop is latched), `computer_waiting` and
++`computer_running`.
++Every request is logged by the Node (`journalctl -u openvibe-node`), with what it would do, approved or denied.
ddiff --git a/docs/plugins.md b/docs/plugins.md
iindex 38e13b8..0aa9157 100644
---- a/docs/plugins.md
++++ b/docs/plugins.md
@@@ -234,23 +234,20 @@ Test it with `openvibe_plugin.Runtime(plugin, infile, outfile, clock)` and a fak
## Where OpenVibe.Actor attaches (later)
--Actor's computer control (terminal, files, screen, keyboard and mouse, browser, the machine's own coding agents) will
--be one more plugin, `computer`, on the same protocol; the core does not change:
--
--- **Capabilities** it will describe: `terminal`, `files`, `screen`, `input`, `browser`, `agents`, each with what the
-- machine allows (for example `{"screen": {"displays": 2}, "agents": {"names": ["claude-code"]}}`).
--- **Command kinds** it will add: `exec`, `fs`, `screen`, `input`, `browser`, `agent`. The core routes by capability
-- key, so new kinds need only a line in `protocol.Kinds` and `plugins.capabilityFor`, plus clamping rules in
-- `safety.Limits.Clamp` if they carry magnitudes. None of them are motion kinds, so deadlines do not apply; they are
-- not in `GuardedKinds`, so an e-stop does not block them unless Actor asks for that.
--- **Control levels** (Observe, Ask, Trusted, Full control) are enforced twice: by Bot/Actor when it issues a command
-- (the operator's `role`), and in the plugin, which reads the owner's local level from its `config` and answers
-- `nack not_allowed` for anything above it. `Ask` makes the plugin hold the command and raise an `event` `approval`
-- that the local UI answers; the command's reply waits (the core's reply timeout is max(deadline, 1 s), so Actor
-- commands will carry a long `deadline_ms` and the cap will come from Actor's own profile).
--- **Output larger than a line** (screenshots, file contents, terminal streams) uses the `video` pattern: the plugin
-- writes a file and sends its path in an `event`, and the core uploads it; the upload path is to be designed with Actor.
--- The local kill switch (`openvibe-node stop`) will also make the `computer` plugin drop to Observe.
++### `computer` — the machine's terminal, files and screen ([computer.md](computer.md))
++
++Actor's computer control (plan T14/T17), on the same protocol. Capabilities `terminal`, `files` and `screen`, each
++with the owner's control level (`observe`, `ask`, `trusted`, `full`) and folders; command kinds `exec`, `fs` and
++`screen` (not motion kinds: long deadlines, no clamping, not in `GuardedKinds`; the core refuses `exec` and file writes
++while a stop is latched). Its `ack` carries the answer in `data`. The level is enforced in the plugin from its local
++config, never from a command; `ask` holds a command (`Deferred` in the SDK), raises an `approval` event and answers
++once the owner runs `openvibe-node approve|deny <id>`, which the core forwards to the plugin as `{"op": "approval", "id",
++"approved"}`. The local kill switch drops it to `observe`. Keyboard and mouse input, a browser and the machine's own
++coding agents are later kinds on the same plugin.
++
++**SDK additions it uses:** `handle()` may return a dict (the ack's `data`) or raise `Deferred` and answer later with
++`complete(id, data)` / `fail(id, code, message)` from any thread; `on_approval(id, approved)`; kinds that move nothing
++keep their full `deadline_ms` (motion kinds stay capped at 5 s).
## Community plugins
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex d60ab96..1e78324 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -173,7 +173,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in
|---------------|-----------------------------------------------------------------------------------------------------|
| `status` | `firmware` (`openvibe-node-<version>`), `capabilities` {per plugin}, `faults` [{`code`, `driver`, `message`}], `estop_latched`; Node extras: `agent_version`, `device_kind`, `os`, `arch`, `drivers` [{`name`, `driver`, `version`, `state`, `capabilities`}], `local_stop`, `video` |
| `telemetry` | `battery` (0..1), `voltage` (V), `rssi`, `sensors` {…}; Node extra: `events` [{`name`, `driver`, `ts`, `fields`}] |
--| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack) |
++| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack); `data` on a computer command (its answer, ≤ 512 KB) |
| `nack` | `id`, `fault_code`; Node extra: `message` |
| `heartbeat` | `t` (send time, Unix ms), `rtt_ms` (the last measured round trip, once there is one; 0 is a real measurement) |
| `reauth` | `token`: a fresh node token for the same principal (a Network-paired machine only; refused → `error` `bot.reauth_refused`) |
@@@ -198,6 +198,9 @@ at most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message
| `button` | `{"name": "forward"}` press; `{"name": …, "state": "down"\|"up"}` hold (names from the robot's profile) | guarded |
| `point` | `{"x": 0..1, "y": 0..1}` normalised click on the video, for the [`relay`](plugins.md) plugin | guarded |
| `halt` | `{}` — stop every actuator now (not latched); always accepted | — |
++| `exec` | `{"argv": ["prog", "arg"…], "cwd"?, "timeout_s"?, "stdin"?}`: run a program on the machine ([computer.md](computer.md)) | computer |
++| `fs` | `{"op": "list"\|"stat"\|"read"\|"write"\|"mkdir", "path", …}`: a file or folder on the machine | computer |
++| `screen` | `{"max_width"?}`: capture the machine's screen | computer |
Sign conventions: `throttle`/`x` positive = forward; `steer` positive = turn right; `y` positive = strafe left;
`rotation` positive = counter-clockwise (left); `pan` positive = right, `tilt` positive = up (plugins can invert a
@@@ -231,6 +234,15 @@ What the Node does with a command, in order:
8. The plugin's `ack`/`nack` is relayed. No answer within max(deadline, 1 s) → the plugin is told to stop and the
server gets `nack plugin_timeout`.
++**Computer kinds** (`exec`, `fs`, `screen`; [computer.md](computer.md)) move nothing. Their `deadline_ms` is the
++reply window (an owner's approval on the machine plus the program's own run): absent → 60 s, capped at 10 minutes, no
++`max_command_ms`. Their value must be a JSON object of at most 384 KB and is not clamped. Running a program and
++writing a file (`exec`, `fs` other than `list`/`stat`/`read`) are refused while the local kill switch is latched
++(`nack local_stop`) or the remote e-stop is (`nack estopped`); capturing the screen and reading files are not. The
++`computer` plugin applies its owner's control level, and its `ack` carries the answer in `data` (at most 512 KB of JSON,
++else `nack bad_value`). An `ask`-level command waits for `openvibe-node approve|deny <id>` on the machine; the plugin
++raises an `approval` event while it waits and `approval_done` when it is answered or expires.
++
Commands are never queued while offline and never replayed after a reconnect.
### Jobs
ddiff --git a/install/install.sh b/install/install.sh
iindex d5f8df7..2edb6db 100755
---- a/install/install.sh
++++ b/install/install.sh
@@@ -24,6 +24,9 @@
# as the unprivileged user openvibe-job in the worker's sandbox with public egress (the service
# runs as root, which run_as and the egress need), and the OpenVibe.Codes job runner is added when
# Node.js 20+ is installed system-wide. Linux only. docs/worker.md has the details.
++# --computer LEVEL this machine's terminal, files and screen for its owner's agents (the computer plugin,
++# docs/computer.md), at LEVEL observe, ask, trusted or full. The service then runs as YOUR account
++# (the one that ran sudo), never root, and its folders are your home. Linux only; not with --worker.
# --no-service install and pair, but do not install the service.
# --local DIR install from DIR (openvibe-node-<os>-<arch> and openvibe-node-plugins.tar.gz) instead of
# downloading.
@@@ -53,13 +56,15 @@ SERVICE=1
LOCAL=""
TAG="latest"
WORKER=0
++COMPUTER=""
# The OpenVibe.Codes release whose `openvibe-codes job` the worker runs (protocol [email protected]).
CODES_VERSION="${OPENVIBE_CODES_VERSION:-v0.3.0}"
CODES_DIR=/opt/openvibe-codes
usage() {
echo "usage: install.sh [--robot rob_… | --network URL --pairing pair_…] [--code CODE | CODE] [--name NAME]"
-- echo " [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none] [--worker] [--no-service]"
++ echo " [--driver adeept|adeept-mecanum|cozmo|looi|ros2|ros2-holonomic|none] [--worker]"
++ echo " [--computer observe|ask|trusted|full] [--no-service]"
echo " [--local DIR] [--version TAG]"
echo " curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept"
echo " curl -fsSL https://openvibe.bot/install | sh -s -- --network https://openvibe.network --pairing pair_… --code ABCD-1234"
@@@ -89,6 +94,8 @@ while [ $# -gt 0 ]; do
--pairing) PAIRING="${2:-}"; shift 2 ;;
--pairing=*) PAIRING="${1#*=}"; shift ;;
--worker) WORKER=1; shift ;;
++ --computer) COMPUTER="${2:-}"; shift 2 ;;
++ --computer=*) COMPUTER="${1#*=}"; shift ;;
--no-service) SERVICE=0; shift ;;
--local) LOCAL="${2:-}"; shift 2 ;;
--version) TAG="${2:-}"; shift 2 ;;
@@@ -127,6 +134,26 @@ fi
# write_worker_config <config.json> <job uid> <job gid> [<node.js> <codes dir>]: turn the job worker on in that config
# (kept as it is otherwise, created when missing): jobs as the job user, public egress, caps sized for a coding task,
# and, with a Node.js and a Codes install, the `[email protected]` job runner. Keys stay empty for the owner to add.
++# write_computer_config <config.json> <level> <home>: add the computer plugin at that level (replacing an earlier entry),
++# its folder the owner's home; the rest of the config is kept.
++write_computer_config() {
++ ${SUDO:-} python3 - "$@" <<'PY'
++import json, os, sys
++path, level, home = sys.argv[1], sys.argv[2], sys.argv[3]
++with open(path) as f:
++ conf = json.load(f)
++plugins = [p for p in conf.get('plugins') or [] if p.get('name') not in ('computer', 'dryrun')]
++plugins.append({'name': 'computer', 'config': {'level': level, 'roots': [home]}})
++conf['plugins'] = plugins
++old = os.umask(0o077)
++with open(path + '.tmp', 'w') as f:
++ json.dump(conf, f, indent=2)
++ f.write('\n')
++os.umask(old)
++os.replace(path + '.tmp', path)
++PY
++}
++
write_worker_config() {
${SUDO:-} python3 - "$@" <<'PY'
import json, os, sys
@@@ -185,6 +212,13 @@ if [ -n "${OPENVIBE_INSTALL_WORKER_CONFIG_ONLY:-}" ]; then
${OPENVIBE_NODEJS:+"$OPENVIBE_NODEJS"} ${OPENVIBE_CODES_DIR:+"$OPENVIBE_CODES_DIR"}
exit 0
fi
++# OPENVIBE_INSTALL_COMPUTER_CONFIG_ONLY=<config.json> with --computer LEVEL and OPENVIBE_COMPUTER_HOME adds the computer
++# plugin to that config and exits (scripts/test-install-computer-config.sh).
++if [ -n "${OPENVIBE_INSTALL_COMPUTER_CONFIG_ONLY:-}" ]; then
++ case "$COMPUTER" in observe|ask|trusted|full) ;; *) die "--computer is observe, ask, trusted or full" ;; esac
++ write_computer_config "$OPENVIBE_INSTALL_COMPUTER_CONFIG_ONLY" "$COMPUTER" "${OPENVIBE_COMPUTER_HOME:?}"
++ exit 0
++fi
# Root for the service, /usr/local/bin and /etc.
SUDO=""
@@@ -353,7 +387,17 @@ say "installed $BIN_DIR/openvibe-node ($("$BIN_DIR/openvibe-node" version))"
# Node's nftables table) need it, and each job still runs as the unprivileged openvibe-job user.
SVC_USER=""
if [ "$WORKER" = 1 ] && [ "$OS" != linux ]; then die "--worker is for Linux"; fi
--if [ "$OS" = linux ] && [ "$SERVICE" = 1 ] && [ "$WORKER" = 0 ]; then
++COMPUTER_HOME=""
++if [ -n "$COMPUTER" ]; then
++ case "$COMPUTER" in observe|ask|trusted|full) ;; *) die "--computer is observe, ask, trusted or full" ;; esac
++ [ "$OS" = linux ] || die "--computer is for Linux in this installer (docs/computer.md: run the Node in your own session elsewhere)"
++ [ "$WORKER" = 0 ] || die "--computer and --worker cannot share a machine yet (the worker's service runs as root)"
++ SVC_USER="${SUDO_USER:-$(id -un)}"
++ [ "$SVC_USER" != root ] || die "--computer runs the Node as your own account: run the installer from it (it uses sudo itself), not as root"
++ COMPUTER_HOME=$(getent passwd "$SVC_USER" | cut -d: -f6)
++ if [ -z "$COMPUTER_HOME" ] || [ ! -d "$COMPUTER_HOME" ]; then die "no home folder for $SVC_USER"; fi
++fi
++if [ "$OS" = linux ] && [ "$SERVICE" = 1 ] && [ "$WORKER" = 0 ] && [ -z "$COMPUTER" ]; then
SVC_USER=openvibe-node
if ! id "$SVC_USER" >/dev/null 2>&1; then
$SUDO useradd --system --home-dir "$STATE" --no-create-home --shell /usr/sbin/nologin "$SVC_USER" 2>/dev/null ||
@@@ -415,6 +459,11 @@ looi)
fi
;;
esac
++if [ -n "$COMPUTER" ]; then
++ # The computer plugin, and Pillow for compact screen captures when pip can get it.
++ $SUDO "$PIP" install --quiet "$STATE/plugins/computer[screen]" 2>/dev/null || $SUDO "$PIP" install --quiet "$STATE/plugins/computer"
++ [ "$DRIVER" = none ] && VIDEO='"source": "off"'
++fi
if [ ! -f "$CONF/config.json" ]; then
$SUDO sh -c "umask 077; cat > '$CONF/config.json'" <<EOF
{
@@@ -428,6 +477,10 @@ EOF
else
say "keeping the existing $CONF/config.json"
fi
++if [ -n "$COMPUTER" ]; then
++ write_computer_config "$CONF/config.json" "$COMPUTER" "$COMPUTER_HOME"
++ say "the computer plugin is on at level $COMPUTER for $SVC_USER (folder $COMPUTER_HOME); change it in $CONF/config.json"
++fi
# ---- the job worker (--worker): the job user, the egress tools, the Codes job runner, the worker config ----
if [ "$WORKER" = 1 ]; then
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex 466868a..02ee3f2 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -342,6 +342,7 @@ var Bundled = map[string]string{
"cozmo": "openvibe_cozmo",
"relay": "openvibe_relay",
"ros2": "openvibe_ros2",
++ "computer": "openvibe_computer",
}
// Default is the config written by `openvibe-node pair` when none exists: the dry-run plugin and its test pattern.
ddiff --git a/internal/fakebot/fakebot.go b/internal/fakebot/fakebot.go
iindex 9302d1b..2849b31 100644
---- a/internal/fakebot/fakebot.go
++++ b/internal/fakebot/fakebot.go
@@@ -546,6 +546,10 @@ func (c *Conn) Reply(id string, timeout time.Duration) (protocol.Message, error)
if m.ID == id {
return m, nil
}
++ case protocol.AckData:
++ if m.ID == id {
++ return m, nil
++ }
case protocol.Nack:
if m.ID == id {
return m, nil
ddiff --git a/internal/localctl/localctl.go b/internal/localctl/localctl.go
iindex 33d7ce6..8256d8b 100644
---- a/internal/localctl/localctl.go
++++ b/internal/localctl/localctl.go
@@@ -1,5 +1,5 @@
--// Package localctl is the running Node's local control socket: `openvibe-node stop`, `resume`, `status` and
--// `plugins` talk to it. It is a Unix domain socket (Windows 10+ supports them too) in the state directory with mode
++// Package localctl is the running Node's local control socket: `openvibe-node stop`, `resume`, `status`, `plugins`,
++// `approvals`, `approve` and `deny` talk to it. It is a Unix domain socket (Windows 10+ supports them too) in the state directory with mode
// 0600, so only the account running the Node and root can use it. One JSON request per connection, one JSON answer.
package localctl
@@@ -19,10 +19,15 @@ const (
CmdStop = "stop"
CmdResume = "resume"
CmdPlugins = "plugins"
++ // The computer plugin's approvals (control level "ask"): list what waits, approve or deny one by its id.
++ CmdApprovals = "approvals"
++ CmdApprove = "approve"
++ CmdDeny = "deny"
)
type Request struct {
Cmd string `json:"cmd"`
++ ID string `json:"id,omitempty"` // approve and deny: the command id the plugin asked about
}
type Response struct {
ddiff --git a/internal/node/computer_test.go b/internal/node/computer_test.go
nnew file mode 100644
iindex 0000000..d1c9390
---- /dev/null
++++ b/internal/node/computer_test.go
@@@ -0,0 +1,238 @@
++//go:build !windows
++
++package node
++
++import (
++ "context"
++ "encoding/json"
++ "io"
++ "log/slog"
++ "os"
++ "os/exec"
++ "path/filepath"
++ "runtime"
++ "strings"
++ "testing"
++ "time"
++
++ "github.com/OpenVibers/OpenVibe.Node/internal/config"
++ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
++ "github.com/OpenVibers/OpenVibe.Node/internal/link"
++ "github.com/OpenVibers/OpenVibe.Node/internal/localctl"
++ "github.com/OpenVibers/OpenVibe.Node/internal/plugins"
++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// startComputer runs a Node with only the real Python computer plugin at the given control level, its roots the one
++// folder it returns, against the fake Bot.
++func startComputer(t *testing.T, level string) (*env, string) {
++ t.Helper()
++ if runtime.GOOS == "windows" {
++ t.Skip("end-to-end tests run on unix")
++ }
++ if os.Geteuid() == 0 {
++ t.Skip("the computer plugin never runs as root")
++ }
++ py, err := exec.LookPath("python3")
++ if err != nil {
++ t.Skip("python3 not installed")
++ }
++ root, _ := filepath.Abs("../..")
++ home, err := os.MkdirTemp("", "ovc")
++ if err != nil {
++ t.Fatal(err)
++ }
++ t.Cleanup(func() { os.RemoveAll(home) })
++ work := filepath.Join(home, "work")
++ if err := os.MkdirAll(work, 0o700); err != nil {
++ t.Fatal(err)
++ }
++ if err := os.WriteFile(filepath.Join(work, "readme.txt"), []byte("hi\n"), 0o600); err != nil {
++ t.Fatal(err)
++ }
++ paths := config.HomePaths(home)
++ srv := fakebot.New()
++ t.Cleanup(srv.Close)
++ srv.Config = protocol.Config{HeartbeatMS: 100, Limits: protocol.Limits{MaxSpeed: f64(0.5), MaxTurn: f64(0.4), MaxCommandMS: 1000},
++ AllowedCommands: append([]string(nil), protocol.Kinds...)}
++ srv.AddCode("TEST2345")
++ creds, err := link.Pair(context.Background(), nil, srv.URL(), protocol.PairRequest{Code: "TEST2345", AgentVersion: "test",
++ DeviceKind: "onboard", Drivers: []string{"computer"}})
++ if err != nil {
++ t.Fatal(err)
++ }
++ cfg := config.Default()
++ cfg.Python = py
++ cfg.Plugins = []config.PluginConfig{{Name: "computer", Config: map[string]any{"level": level, "roots": []string{work}, "timeout_s": 20},
++ Env: map[string]string{"PYTHONPATH": filepath.Join(root, "plugins/sdk") + string(os.PathListSeparator) + filepath.Join(root, "plugins/computer")}}}
++ cfg.Video = config.VideoConfig{Source: "off"}
++ n, err := New(Options{Config: cfg, Paths: paths, Creds: creds, Version: "test", HeartbeatInterval: 100 * time.Millisecond,
++ LinkBackoffMin: 50 * time.Millisecond, LatchPoll: 50 * time.Millisecond, Log: testLog()})
++ if err != nil {
++ t.Fatal(err)
++ }
++ ctx, cancel := context.WithCancel(context.Background())
++ done := make(chan struct{})
++ go func() { n.Run(ctx); close(done) }()
++ e := &env{t: t, srv: srv, node: n, paths: paths, cancel: cancel, done: done}
++ t.Cleanup(e.stop)
++ e.conn = e.nextConn()
++ if _, err := e.conn.Expect(protocol.TypeStatus, 20*time.Second, func(m protocol.Message) bool {
++ s := m.(protocol.Status)
++ if os.Getenv("OV_DEBUG") != "" {
++ b, _ := json.Marshal(s.Drivers)
++ t.Logf("status drivers %s", b)
++ }
++ return len(s.Drivers) == 1 && s.Drivers[0].State == plugins.StateReady
++ }); err != nil {
++ t.Fatal(err)
++ }
++ real, _ := filepath.EvalSymlinks(work)
++ return e, real
++}
++
++// computerCmd sends a computer command with a 20 s deadline and returns its reply (waiting up to wait).
++func (e *env) computerCmd(id, kind string, value any, wait time.Duration) protocol.Message {
++ e.t.Helper()
++ b, _ := json.Marshal(value)
++ if err := e.conn.Command(protocol.Command{ID: id, Kind: kind, Value: b}, 20000); err != nil {
++ e.t.Fatal(err)
++ }
++ r, err := e.conn.Reply(id, wait)
++ if err != nil {
++ e.t.Fatal(err)
++ }
++ return r
++}
++
++func testLog() *slog.Logger {
++ if os.Getenv("OV_DEBUG") != "" {
++ return slog.New(slog.NewTextHandler(os.Stderr, nil))
++ }
++ return slog.New(slog.NewTextHandler(io.Discard, nil))
++}
++
++func dataOf(t *testing.T, m protocol.Message) map[string]any {
++ t.Helper()
++ a, ok := m.(protocol.AckData)
++ if !ok {
++ t.Fatalf("want an ack with data, got %#v", m)
++ }
++ var d map[string]any
++ if err := json.Unmarshal(a.Data, &d); err != nil {
++ t.Fatal(err)
++ }
++ return d
++}
++
++func nackOf(t *testing.T, m protocol.Message) protocol.Nack {
++ t.Helper()
++ nk, ok := m.(protocol.Nack)
++ if !ok {
++ t.Fatalf("want a nack, got %#v", m)
++ }
++ return nk
++}
++
++func TestComputerTrustedRunsAndAnswersWithData(t *testing.T) {
++ e, work := startComputer(t, "trusted")
++ d := dataOf(t, e.computerCmd("x1", protocol.KindExec, map[string]any{"argv": []string{"python3", "-c", "import os; print(os.getcwd())"}}, 15*time.Second))
++ if strings.TrimSpace(d["stdout"].(string)) != work || d["exit_code"].(float64) != 0 {
++ t.Fatalf("exec: %v", d)
++ }
++ d = dataOf(t, e.computerCmd("x2", protocol.KindFS, map[string]any{"op": "list", "path": work}, 5*time.Second))
++ if entries := d["entries"].([]any); len(entries) != 1 || entries[0].(map[string]any)["name"] != "readme.txt" {
++ t.Fatalf("list: %v", d)
++ }
++ if nk := nackOf(t, e.computerCmd("x3", protocol.KindFS, map[string]any{"op": "read", "path": "/etc/hostname"}, 5*time.Second)); nk.FaultCode != protocol.FaultNotAllowed {
++ t.Fatalf("outside the roots: %+v", nk)
++ }
++ if nk := nackOf(t, e.computerCmd("x4", protocol.KindExec, "rm -rf /", 5*time.Second)); nk.FaultCode != protocol.FaultBadValue {
++ t.Fatalf("a value that is not an object: %+v", nk)
++ }
++}
++
++func TestComputerAskWaitsForTheOwnerOnTheMachine(t *testing.T) {
++ e, _ := startComputer(t, "ask")
++ b, _ := json.Marshal(map[string]any{"argv": []string{"echo", "approved"}})
++ if err := e.conn.Command(protocol.Command{ID: "a1", Kind: protocol.KindExec, Value: b}, 20000); err != nil {
++ t.Fatal(err)
++ }
++ var list []Approval
++ deadline := time.Now().Add(10 * time.Second)
++ for len(list) == 0 && time.Now().Before(deadline) {
++ r, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdApprovals}, 2*time.Second)
++ if err != nil || !r.OK {
++ t.Fatal(err, r.Error)
++ }
++ _ = json.Unmarshal(r.Data, &list)
++ if len(list) == 0 {
++ time.Sleep(50 * time.Millisecond)
++ }
++ }
++ if len(list) != 1 || list[0].ID != "a1" || list[0].Kind != protocol.KindExec || !strings.Contains(list[0].Summary, "echo approved") {
++ t.Fatalf("approvals: %+v", list)
++ }
++ if r, _ := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdApprove, ID: "nope"}, 2*time.Second); r.OK {
++ t.Fatal("an unknown id was approved")
++ }
++ if r, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdApprove, ID: "a1"}, 2*time.Second); err != nil || !r.OK {
++ t.Fatal(err, r.Error)
++ }
++ r, err := e.conn.Reply("a1", 10*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if d := dataOf(t, r); d["stdout"] != "approved\n" {
++ t.Fatalf("approved exec: %v", d)
++ }
++ if got := e.node.Approvals(); len(got) != 0 {
++ t.Fatalf("still listed after its answer: %+v", got)
++ }
++ if err := e.conn.Command(protocol.Command{ID: "a2", Kind: protocol.KindExec, Value: b}, 20000); err != nil {
++ t.Fatal(err)
++ }
++ for time.Now().Before(deadline.Add(5*time.Second)) && len(e.node.Approvals()) == 0 {
++ time.Sleep(50 * time.Millisecond)
++ }
++ if r, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdDeny, ID: "a2"}, 2*time.Second); err != nil || !r.OK {
++ t.Fatal(err, r.Error)
++ }
++ r, err = e.conn.Reply("a2", 10*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if nk := nackOf(t, r); nk.FaultCode != protocol.FaultNotAllowed {
++ t.Fatalf("denied: %+v", nk)
++ }
++}
++
++func TestComputerLocalStopLeavesObserveOnly(t *testing.T) {
++ e, work := startComputer(t, "trusted")
++ if r, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStop}, 2*time.Second); err != nil || !r.OK {
++ t.Fatal(err, r.Error)
++ }
++ if nk := nackOf(t, e.computerCmd("s1", protocol.KindExec, map[string]any{"argv": []string{"true"}}, 5*time.Second)); nk.FaultCode != protocol.FaultLocalStop {
++ t.Fatalf("exec while stopped: %+v", nk)
++ }
++ if nk := nackOf(t, e.computerCmd("s2", protocol.KindFS, map[string]any{"op": "write", "path": filepath.Join(work, "x"), "content_b64": "eA=="}, 5*time.Second)); nk.FaultCode != protocol.FaultLocalStop {
++ t.Fatalf("write while stopped: %+v", nk)
++ }
++ if d := dataOf(t, e.computerCmd("s3", protocol.KindFS, map[string]any{"op": "stat", "path": work}, 5*time.Second)); d["type"] != "dir" {
++ t.Fatalf("observing while stopped: %v", d)
++ }
++ if r, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdResume}, 2*time.Second); err != nil || !r.OK {
++ t.Fatal(err, r.Error)
++ }
++ time.Sleep(200 * time.Millisecond) // the plugin hears the resume
++ if d := dataOf(t, e.computerCmd("s4", protocol.KindExec, map[string]any{"argv": []string{"echo", "back"}}, 10*time.Second)); d["stdout"] != "back\n" {
++ t.Fatalf("after resume: %v", d)
++ }
++}
++
++func TestComputerKindsNeedTheComputerPlugin(t *testing.T) {
++ e := start(t)
++ if nk := nackOf(t, e.cmd(protocol.KindExec, `{"argv":["true"]}`, 5000)); nk.FaultCode != protocol.FaultUnsupported {
++ t.Fatalf("exec without the computer plugin: %+v", nk)
++ }
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex b12ea5f..0e2d93c 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -86,6 +86,20 @@ type Node struct {
configured bool
robotIDs []string // the robots this device serves, from the connection's hello
++
++ // approvals waiting on the machine (the computer plugin at level "ask"), by command id: what
++ // `openvibe-node approvals` lists and `approve`/`deny` answer.
++ approvals map[string]Approval
++}
++
++// Approval is one computer command waiting for its owner's answer on the machine.
++type Approval struct {
++ ID string `json:"id"`
++ Plugin string `json:"plugin"`
++ Kind string `json:"kind,omitempty"`
++ Summary string `json:"summary,omitempty"`
++ At int64 `json:"at"`
++ ExpiresS int64 `json:"expires_s,omitempty"`
}
// New builds a Node. Run starts it.
@@@ -101,7 +115,7 @@ func New(opt Options) (*Node, error) {
opt.Log.Error("latch", "err", err)
}
n := &Node{opt: opt, log: opt.Log, latch: latch, dedup: safety.NewDedup(4096, 10*time.Minute),
-- jobs: newJobs(nil), telemetry: map[string]map[string]any{}, telemMS: 500}
++ jobs: newJobs(nil), telemetry: map[string]map[string]any{}, telemMS: 500, approvals: map[string]Approval{}}
n.limits = safety.Merge(protocol.Limits{}, opt.Config.Limits.MaxSpeed, opt.Config.Limits.MaxTurn, opt.Config.Limits.MaxCommandMS)
var specs []plugins.Spec
@@@ -393,6 +407,10 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
nack(protocol.FaultNotAllowed, c.Kind+" is not in this robot's allowed_commands")
return
}
++ if protocol.ComputerKinds[c.Kind] {
++ n.computer(c, entry, sentMS, received, nack)
++ return
++ }
if protocol.GuardedKinds[c.Kind] {
st := n.latch.State()
if st.Local {
@@@ -450,6 +468,64 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
}()
}
++// computer runs one exec, fs or screen command: no clamping, a long reply window (the server's deadline, at most
++// MaxComputerMS, else DefaultComputerMS), refused while the local stop or an e-stop is latched when it would change the
++// machine, and answered with the plugin's data. The plugin applies its owner's control level.
++func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, received time.Time, nack func(code, msg string)) {
++ if len(c.Value) > protocol.MaxComputerValue {
++ nack(protocol.FaultBadValue, fmt.Sprintf("the value is larger than %d KB", protocol.MaxComputerValue/1024))
++ return
++ }
++ var obj map[string]json.RawMessage
++ if json.Unmarshal(c.Value, &obj) != nil || obj == nil {
++ nack(protocol.FaultBadValue, "value must be a JSON object")
++ return
++ }
++ if protocol.ComputerActs(c.Kind, c.Value) {
++ st := n.latch.State()
++ if st.Local {
++ nack(protocol.FaultLocalStop, "stopped on the machine with `openvibe-node stop`: observe only")
++ return
++ }
++ if st.Remote {
++ nack(protocol.FaultEstopped, st.RemoteReason)
++ return
++ }
++ }
++ window := int64(protocol.DefaultComputerMS)
++ if c.DeadlineMS > 0 {
++ window = c.DeadlineMS - n.serverNow(sentMS, received)
++ if window <= 0 {
++ nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived", -window))
++ return
++ }
++ window = min(window, int64(protocol.MaxComputerMS))
++ }
++ p := n.mgr.Route(c.Kind, c.Target, "")
++ if p == nil {
++ nack(protocol.FaultUnsupported, "no driver on this device handles "+c.Kind+" (the computer plugin is off)")
++ return
++ }
++ wait := p.Begin(n.ctx, c.ID, c.Kind, c.Value, int(window))
++ go func() {
++ r := wait()
++ var msg protocol.Message
++ switch {
++ case r.OK && len(r.Data) > protocol.MaxAckData:
++ msg = protocol.Nack{ID: c.ID, FaultCode: protocol.FaultBadValue, Message: fmt.Sprintf("the answer is larger than %d KB", protocol.MaxAckData/1024)}
++ case r.OK:
++ msg = protocol.AckData{ID: c.ID, LatencyMS: time.Since(received).Milliseconds(), Data: r.Data}
++ default:
++ msg = protocol.Nack{ID: c.ID, FaultCode: r.FaultCode, Message: r.Message}
++ }
++ n.mu.Lock()
++ delete(n.approvals, c.ID)
++ n.mu.Unlock()
++ n.dedup.Complete(entry, msg)
++ n.send(msg)
++ }()
++}
++
// serverNow is the server's clock when a frame stamped sentMS was received: the link's estimate, else the frame's
// own ts.
func (n *Node) serverNow(sentMS int64, received time.Time) int64 {
@@@ -566,6 +642,7 @@ func (n *Node) output(o plugins.Output) {
fields = nil
}
n.log.Info("plugin event", "plugin", o.Plugin, "event", name)
++ n.trackApproval(o.Plugin, name, o.Data, o.At)
n.mu.Lock()
n.events = append(n.events, protocol.Event{Name: name, Driver: o.Plugin, TS: o.At.UnixMilli(), Fields: fields})
n.mu.Unlock()
@@@ -873,8 +950,68 @@ func (n *Node) sendStatus() {
n.send(s)
}
++// trackApproval keeps the list `openvibe-node approvals` shows from the computer plugin's approval events.
++func (n *Node) trackApproval(plugin, name string, data map[string]any, at time.Time) {
++ id, _ := data["id"].(string)
++ if id == "" || (name != "approval" && name != "approval_done") {
++ return
++ }
++ n.mu.Lock()
++ defer n.mu.Unlock()
++ if name == "approval_done" {
++ delete(n.approvals, id)
++ return
++ }
++ if len(n.approvals) >= 64 {
++ return
++ }
++ a := Approval{ID: id, Plugin: plugin, At: at.UnixMilli()}
++ a.Kind, _ = data["kind"].(string)
++ a.Summary, _ = data["summary"].(string)
++ if e, ok := data["expires_s"].(float64); ok {
++ a.ExpiresS = int64(e)
++ }
++ n.approvals[id] = a
++}
++
++// Approvals lists the computer commands waiting for an answer on the machine, oldest first.
++func (n *Node) Approvals() []Approval {
++ n.mu.Lock()
++ defer n.mu.Unlock()
++ out := make([]Approval, 0, len(n.approvals))
++ for _, a := range n.approvals {
++ out = append(out, a)
++ }
++ sort.Slice(out, func(i, j int) bool { return out[i].At < out[j].At || (out[i].At == out[j].At && out[i].ID < out[j].ID) })
++ return out
++}
++
func (n *Node) control(r localctl.Request) localctl.Response {
switch r.Cmd {
++ case localctl.CmdApprovals:
++ b, err := json.Marshal(n.Approvals())
++ if err != nil {
++ return localctl.Response{Error: err.Error()}
++ }
++ return localctl.Response{OK: true, Data: b}
++ case localctl.CmdApprove, localctl.CmdDeny:
++ n.mu.Lock()
++ a, ok := n.approvals[r.ID]
++ if ok {
++ delete(n.approvals, r.ID)
++ }
++ n.mu.Unlock()
++ if !ok {
++ return localctl.Response{Error: fmt.Sprintf("nothing with the id %q waits for approval", strings.TrimSpace(r.ID))}
++ }
++ p := n.mgr.Get(a.Plugin)
++ if p == nil {
++ return localctl.Response{Error: "the plugin that asked is gone"}
++ }
++ if err := p.Answer(a.ID, r.Cmd == localctl.CmdApprove); err != nil {
++ return localctl.Response{Error: err.Error()}
++ }
++ return localctl.Response{OK: true}
case localctl.CmdStop:
n.mgr.EstopAll() // at once, then persist
if _, err := n.latch.SetLocalWith("openvibe-node stop"); err != nil {
ddiff --git a/internal/plugins/manager.go b/internal/plugins/manager.go
iindex 2144199..be2d7a0 100644
---- a/internal/plugins/manager.go
++++ b/internal/plugins/manager.go
@@@ -17,6 +17,9 @@ var capabilityFor = map[string]string{
protocol.KindDisplay: "display",
protocol.KindButton: "button",
protocol.KindPoint: "point",
++ protocol.KindExec: "terminal",
++ protocol.KindFS: "files",
++ protocol.KindScreen: "screen",
}
// Manager supervises every configured plugin.
ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
iindex 802d352..da60582 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -64,6 +64,7 @@ type Reply struct {
OK bool
FaultCode string
Message string
++ Data json.RawMessage // an ack's data (a computer command's answer), as the plugin sent it
}
// Output kinds carried by Output.Kind.
@@@ -200,6 +201,11 @@ func (p *Plugin) Resume() {
_ = p.send(map[string]any{"op": "resume"})
}
++// Answer gives the plugin its owner's answer to an approval it asked for (`openvibe-node approve|deny <id>`).
++func (p *Plugin) Answer(id string, approved bool) error {
++ return p.send(map[string]any{"op": "approval", "id": id, "approved": approved})
++}
++
// Command sends a command and waits for the plugin's ack or nack.
func (p *Plugin) Command(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int) Reply {
return p.Begin(ctx, id, kind, value, deadlineMS)()
@@@ -503,12 +509,16 @@ func (p *Plugin) readStdout(r io.Reader, described chan struct{}) {
id, _ := m["id"].(string)
code, _ := m["fault_code"].(string)
msg, _ := m["message"].(string)
++ var data json.RawMessage
++ if d, ok := m["data"]; ok && op == "ack" && d != nil {
++ data, _ = json.Marshal(d)
++ }
p.mu.Lock()
ch := p.pending[id]
p.mu.Unlock()
if ch != nil {
select {
-- case ch <- Reply{OK: op == "ack", FaultCode: code, Message: msg}:
++ case ch <- Reply{OK: op == "ack", FaultCode: code, Message: msg, Data: data}:
default:
}
}
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 4572c02..9d24877 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -155,10 +155,45 @@ const (
KindHalt = "halt"
KindButton = "button"
KindPoint = "point"
++
++ // The machine's own computer (the `computer` plugin, docs/plugins.md): run a program, use a file, capture the
++ // screen. Their ack carries the answer in data; the owner's control level on the machine decides what runs.
++ KindExec = "exec"
++ KindFS = "fs"
++ KindScreen = "screen"
)
// Kinds lists every command kind the Node understands.
--var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt, KindButton, KindPoint}
++var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt, KindButton, KindPoint, KindExec, KindFS, KindScreen}
++
++// ComputerKinds move nothing: they carry long deadlines (DefaultComputerMS, at most MaxComputerMS) and no clamping.
++var ComputerKinds = map[string]bool{KindExec: true, KindFS: true, KindScreen: true}
++
++const (
++ // DefaultComputerMS is a computer command's reply window when the server sends no deadline; MaxComputerMS caps
++ // one (an approval on the machine plus the program's own run).
++ DefaultComputerMS = 60_000
++ MaxComputerMS = 600_000
++ // MaxComputerValue caps a computer command's value (a file write's content); MaxAckData an ack's data.
++ MaxComputerValue = 384 * 1024
++ MaxAckData = 512 * 1024
++)
++
++// ComputerActs reports whether a computer command changes the machine (runs a program, writes a file): those are
++// refused while the local stop or an e-stop is latched. Capturing the screen and reading files are not.
++func ComputerActs(kind string, value json.RawMessage) bool {
++ switch kind {
++ case KindExec:
++ return true
++ case KindFS:
++ var v struct {
++ Op string `json:"op"`
++ }
++ _ = json.Unmarshal(value, &v)
++ return v.Op != "list" && v.Op != "stat" && v.Op != "read"
++ }
++ return false
++}
// GuardedKinds move something and are refused while an e-stop or the local kill switch is latched.
var GuardedKinds = map[string]bool{KindDrive: true, KindActuator: true, KindPTZ: true, KindButton: true, KindPoint: true}
@@@ -527,6 +562,14 @@ type Ack struct {
LatencyMS int64 `json:"latency_ms,omitempty"`
}
++// AckData is the ack of a computer command: an ack frame that also carries the command's answer (exec output, a file,
++// a screen), at most MaxAckData bytes of JSON.
++type AckData struct {
++ ID string `json:"id"`
++ LatencyMS int64 `json:"latency_ms,omitempty"`
++ Data json.RawMessage `json:"data,omitempty"`
++}
++
type Nack struct {
ID string `json:"id"`
FaultCode string `json:"fault_code"`
@@@ -566,6 +609,7 @@ func (Error) MessageType() string { return TypeError }
func (Status) MessageType() string { return TypeStatus }
func (Telemetry) MessageType() string { return TypeTelemetry }
func (Ack) MessageType() string { return TypeAck }
++func (AckData) MessageType() string { return TypeAck }
func (Nack) MessageType() string { return TypeNack }
func (Heartbeat) MessageType() string { return TypeHeartbeat }
func (Reauth) MessageType() string { return TypeReauth }
@@@ -587,7 +631,7 @@ var registry = map[string]func() Message{
TypeError: func() Message { return &Error{} },
TypeStatus: func() Message { return &Status{} },
TypeTelemetry: func() Message { return &Telemetry{} },
-- TypeAck: func() Message { return &Ack{} },
++ TypeAck: func() Message { return &AckData{} }, // an Ack unless it carries data
TypeNack: func() Message { return &Nack{} },
TypeHeartbeat: func() Message { return &Heartbeat{} },
TypeReauth: func() Message { return &Reauth{} },
@@@ -653,6 +697,11 @@ func Decode(data []byte) (Frame, error) {
func derefMessage(m Message) Message {
switch v := m.(type) {
++ case *AckData:
++ if len(v.Data) == 0 {
++ return Ack{ID: v.ID, LatencyMS: v.LatencyMS}
++ }
++ return *v
case *Hello:
return *v
case *Config:
ddiff --git a/plugins/computer/openvibe_computer/__init__.py b/plugins/computer/openvibe_computer/__init__.py
nnew file mode 100644
iindex 0000000..fabdace
---- /dev/null
++++ b/plugins/computer/openvibe_computer/__init__.py
@@@ -0,0 +1,527 @@
++"""OpenVibe Node computer plugin: the machine's terminal, files and screen for its owner's agents (plan T14/T17).
++
++Off unless its owner lists it in the Node's config. The owner sets the **control level** on the machine itself; a cloud
++command can never raise it:
++
++* ``observe`` (the default): capture the screen; list, stat and read files inside the roots.
++* ``ask``: also run programs and write files inside the roots, each only after the owner approves it on the machine
++ (``openvibe-node approvals``, ``approve <id>``, ``deny <id>``); unanswered requests expire.
++* ``trusted``: run programs and write files inside the roots without asking.
++* ``full``: as trusted, anywhere the account running the Node can reach.
++
++The local kill switch (``openvibe-node stop``) and a remote e-stop drop the plugin to observe, kill what is running
++and deny what waits; the core refuses acting commands while either is latched too. The plugin never runs as root.
++
++Command kinds and their answers (the ack's ``data``):
++
++* ``exec`` ``{"argv": [...], "cwd"?, "timeout_s"?, "stdin"?}`` → ``{exit_code, stdout, stderr, truncated, timed_out,
++ duration_ms}``. A program and its arguments, never a shell line; output is cut at ``max_output_bytes``.
++* ``fs`` ``{"op": "list"|"stat"|"read"|"write"|"mkdir", "path", …}``: ``read`` takes ``offset``/``length`` and answers
++ ``content_b64``; ``write`` takes ``content_b64`` and ``mode`` (``create``, ``overwrite`` or ``append``).
++* ``screen`` ``{"max_width"?}`` → ``{format, width, height, content_b64}`` (JPEG through Pillow when it is installed).
++"""
++
++import base64
++import os
++import shlex
++import shutil
++import stat
++import subprocess
++import sys
++import tempfile
++import threading
++import time
++
++from openvibe_plugin import Deferred, Fault, Plugin
++
++LEVELS = ("observe", "ask", "trusted", "full")
++RANK = {name: i for i, name in enumerate(LEVELS)}
++READ_OPS = frozenset({"list", "stat", "read"})
++WRITE_OPS = frozenset({"write", "mkdir"})
++MAX_ARGS = 64
++MAX_ARG_LEN = 4096
++MAX_STDIN = 64 * 1024
++MAX_LIST = 1000
++MAX_RUNNING = 2
++MAX_WAITING = 8
++# The core caps an ack's data at 512 KB of JSON; base64 grows by a third.
++MAX_OUTPUT_CAP = 192 * 1024
++MAX_READ_CAP = 256 * 1024
++SCREEN_MAX_BYTES = 300 * 1024
++
++
++def _int(v, name, lo, hi, default):
++ if v is None:
++ return default
++ if isinstance(v, bool) or not isinstance(v, (int, float)) or v != v:
++ raise Fault("bad_value", "%s must be a number" % name)
++ return int(max(lo, min(hi, v)))
++
++
++class Computer(Plugin):
++ driver = "computer"
++ version = "0.1.0"
++ motion_kinds = frozenset()
++ telemetry_interval_s = 2.0
++
++ def __init__(self, which=shutil.which, popen=subprocess.Popen, clock=time.monotonic, env=None, platform=None):
++ self.which = which
++ self.popen = popen
++ self.clock = clock
++ self.env = os.environ if env is None else env
++ self.platform = platform or sys.platform
++ self.level = "observe"
++ self.roots = []
++ self.timeout_s = 30
++ self.max_output = 64 * 1024
++ self.max_read = 256 * 1024
++ self.allow_exec = True
++ self.approval_timeout_s = 120
++ self.pending = {} # command id -> (run, expires_at, summary)
++ self.procs = set()
++ self.running = 0
++ self.mu = threading.Lock()
++
++ # ---- configuration -------------------------------------------------------------------------------------------
++ def _configure(self, config):
++ level = str(config.get("level") or "observe").strip().lower()
++ if level not in RANK:
++ raise Fault("bad_value", "level is one of %s" % ", ".join(LEVELS))
++ roots = config.get("roots")
++ if roots is None:
++ roots = [os.path.expanduser("~")]
++ if not isinstance(roots, list) or not all(isinstance(r, str) and r.strip() for r in roots):
++ raise Fault("bad_value", "roots is a list of folders")
++ self.level = level
++ self.roots = [os.path.realpath(os.path.expanduser(r.strip())) for r in roots]
++ self.timeout_s = _int(config.get("timeout_s"), "timeout_s", 1, 600, 30)
++ self.max_output = _int(config.get("max_output_bytes"), "max_output_bytes", 1024, MAX_OUTPUT_CAP, 64 * 1024)
++ self.max_read = _int(config.get("max_read_bytes"), "max_read_bytes", 1024, MAX_READ_CAP, 256 * 1024)
++ self.approval_timeout_s = _int(config.get("approval_timeout_s"), "approval_timeout_s", 10, 600, 120)
++ self.allow_exec = config.get("allow_exec", True) is not False
++
++ def describe(self, config):
++ try:
++ self._configure(config)
++ except Fault:
++ return {}
++ caps = {"files": {"level": self.level, "roots": self.roots, "max_read_bytes": self.max_read}}
++ if self.allow_exec:
++ caps["terminal"] = {"level": self.level, "timeout_s": self.timeout_s, "max_output_bytes": self.max_output}
++ tool = self._screen_tool()
++ if tool:
++ caps["screen"] = {"level": self.level, "tool": tool[0]}
++ return caps
++
++ def setup(self, config):
++ self._configure(config)
++ if hasattr(os, "geteuid") and os.geteuid() == 0:
++ raise Fault("not_allowed", "the computer plugin never runs as root: run the Node as your own account")
++
++ # ---- safety ----------------------------------------------------------------------------------------------------
++ def effective_level(self):
++ """The level now: observe while a local stop or an e-stop is latched."""
++ if self.runtime is not None and self.runtime.estopped:
++ return "observe"
++ return self.level
++
++ def stop(self):
++ """halt, stop, the local kill switch and an e-stop: kill what runs, deny what waits."""
++ with self.mu:
++ procs = list(self.procs)
++ waiting = list(self.pending.keys())
++ self.pending.clear()
++ for p in procs:
++ self._kill(p)
++ for cid in waiting:
++ self.emit_event("approval_done", id=cid, approved=False, reason="stopped")
++ self.fail(cid, "local_stop", "stopped on the machine before it ran")
++
++ def _kill(self, proc):
++ try:
++ if hasattr(os, "killpg"):
++ os.killpg(proc.pid, 9)
++ else:
++ proc.kill()
++ except (OSError, ProcessLookupError):
++ pass
++
++ def poll(self):
++ now = self.clock()
++ expired = []
++ with self.mu:
++ for cid, (_, at, _) in list(self.pending.items()):
++ if now >= at:
++ expired.append(cid)
++ del self.pending[cid]
++ waiting, running = len(self.pending), self.running
++ for cid in expired:
++ self.emit_event("approval_done", id=cid, approved=False, reason="expired")
++ self.fail(cid, "not_allowed", "nobody answered on the machine in time")
++ # Sensors are what the core forwards to Bot: the machines page shows the level and what waits on the machine.
++ return {"sensors": {"computer_level": self.effective_level(), "computer_waiting": waiting, "computer_running": running}}
++
++ # ---- paths -----------------------------------------------------------------------------------------------------
++ def _inside(self, path):
++ return any(path == r or path.startswith(r.rstrip(os.sep) + os.sep) for r in self.roots)
++
++ def _path(self, raw, level, create=False):
++ """The real path a command names, inside the roots unless the level is full."""
++ if not isinstance(raw, str) or not raw or len(raw) > MAX_ARG_LEN or "\x00" in raw:
++ raise Fault("bad_value", "path is a non-empty string")
++ path = os.path.expanduser(raw)
++ if not os.path.isabs(path):
++ raise Fault("bad_value", "path must be absolute")
++ if create and not os.path.lexists(path):
++ parent = os.path.realpath(os.path.dirname(path))
++ real = os.path.join(parent, os.path.basename(path))
++ else:
++ real = os.path.realpath(path)
++ if level != "full" and not self._inside(real):
++ raise Fault("not_allowed", "%s is outside this machine's allowed folders" % real)
++ return real
++
++ # ---- commands --------------------------------------------------------------------------------------------------
++ def handle(self, cmd):
++ v = cmd.value
++ if not isinstance(v, dict):
++ raise Fault("bad_value", "value must be an object")
++ level = self.effective_level()
++ if cmd.kind == "screen":
++ tool = self._screen_tool()
++ if not tool:
++ raise Fault("unsupported", "no screen capture tool on this machine")
++ width = _int(v.get("max_width"), "max_width", 320, 3840, 1280)
++ return self._start(cmd, lambda: self._screen(tool, width))
++ if cmd.kind == "fs":
++ op = v.get("op")
++ if op in READ_OPS:
++ return self._fs_read_op(op, v, level)
++ if op not in WRITE_OPS:
++ raise Fault("bad_value", "op is list, stat, read, write or mkdir")
++ run, summary = self._fs_write_op(op, v, level)
++ return self._act(cmd, level, run, summary, threaded=False)
++ if cmd.kind == "exec":
++ if not self.allow_exec:
++ raise Fault("unsupported", "this machine does not allow running programs")
++ run, summary = self._exec_op(cmd, v, level)
++ return self._act(cmd, level, run, summary, threaded=True)
++ raise Fault("unsupported", "kind %s not supported" % cmd.kind)
++
++ def _act(self, cmd, level, run, summary, threaded):
++ if RANK[level] < RANK["ask"]:
++ stopped = self.runtime is not None and self.runtime.estopped
++ raise Fault("local_stop" if stopped else "not_allowed",
++ "stopped on the machine: observe only" if stopped else
++ "this machine allows observing only (its owner sets the level on the machine)")
++ if level == "ask":
++ with self.mu:
++ if len(self.pending) >= MAX_WAITING:
++ raise Fault("not_ready", "%d requests already wait for approval on the machine" % MAX_WAITING)
++ self.pending[cmd.id] = ((run, threaded), self.clock() + self.approval_timeout_s, summary)
++ self.emit_event("approval", id=cmd.id, kind=cmd.kind, summary=summary, expires_s=self.approval_timeout_s)
++ self.log("waiting for approval %s: %s", cmd.id, summary)
++ raise Deferred()
++ self.log("%s %s: %s", cmd.kind, cmd.id, summary)
++ return self._start(cmd, run) if threaded else run()
++
++ def on_approval(self, cid, approved):
++ with self.mu:
++ item = self.pending.pop(cid, None)
++ if item is None:
++ return
++ (run, threaded), _, summary = item
++ self.emit_event("approval_done", id=cid, approved=bool(approved))
++ if not approved:
++ self.log("denied %s: %s", cid, summary)
++ self.fail(cid, "not_allowed", "the owner denied it on the machine")
++ return
++ if self.effective_level() == "observe":
++ self.fail(cid, "local_stop", "stopped on the machine before it ran")
++ return
++ self.log("approved %s: %s", cid, summary)
++ if threaded:
++ self._thread(cid, run)
++ return
++ try:
++ self.complete(cid, run())
++ except Fault as f:
++ self.fail(cid, f.code, f.message)
++ except Exception as e:
++ self.fail(cid, "hardware", str(e))
++
++ def _start(self, cmd, run):
++ with self.mu:
++ if self.running >= MAX_RUNNING:
++ raise Fault("not_ready", "%d commands are already running on this machine" % MAX_RUNNING)
++ self._thread(cmd.id, run)
++ raise Deferred()
++
++ def _thread(self, cid, run):
++ with self.mu:
++ self.running += 1
++
++ def work():
++ try:
++ self.complete(cid, run())
++ except Fault as f:
++ self.fail(cid, f.code, f.message)
++ except Exception as e:
++ self.fail(cid, "hardware", str(e))
++ finally:
++ with self.mu:
++ self.running -= 1
++
++ threading.Thread(target=work, name="computer-%s" % cid, daemon=True).start()
++
++ # ---- exec ------------------------------------------------------------------------------------------------------
++ def _exec_op(self, cmd, v, level):
++ argv = v.get("argv")
++ if not isinstance(argv, list) or not argv or len(argv) > MAX_ARGS or \
++ not all(isinstance(a, str) and len(a) <= MAX_ARG_LEN and "\x00" not in a for a in argv) or not argv[0]:
++ raise Fault("bad_value", "argv is a list of 1 to %d strings: a program and its arguments" % MAX_ARGS)
++ cwd = self._path(v["cwd"], level) if v.get("cwd") is not None else (self.roots[0] if self.roots else None)
++ if cwd is not None and not os.path.isdir(cwd):
++ raise Fault("bad_value", "cwd %s is not a folder" % cwd)
++ stdin = v.get("stdin")
++ if stdin is not None and (not isinstance(stdin, str) or len(stdin.encode("utf-8")) > MAX_STDIN):
++ raise Fault("bad_value", "stdin is a string of at most 64 KB")
++ timeout = _int(v.get("timeout_s"), "timeout_s", 1, self.timeout_s, self.timeout_s)
++ deadline = cmd.deadline
++
++ def run():
++ left = deadline - self.clock() - 1.0
++ if left < 1:
++ raise Fault("expired", "no time was left to run it")
++ return self._exec(argv, cwd, stdin, min(timeout, left))
++
++ return run, "run %s in %s" % (shlex.join(argv)[:300], cwd)
++
++ def _exec(self, argv, cwd, stdin, timeout):
++ started = self.clock()
++ try:
++ proc = self.popen(argv, cwd=cwd, stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
++ stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True)
++ except FileNotFoundError:
++ raise Fault("bad_value", "no program %s on this machine" % argv[0])
++ except PermissionError:
++ raise Fault("not_allowed", "%s may not be run by the Node's account" % argv[0])
++ with self.mu:
++ self.procs.add(proc)
++ cap = self.max_output
++ got = {"out": [b"", 0], "err": [b"", 0]}
++
++ def drain(stream, key):
++ buf = got[key]
++ while True:
++ chunk = stream.read(65536)
++ if not chunk:
++ return
++ buf[1] += len(chunk)
++ if len(buf[0]) < cap:
++ buf[0] += chunk[:cap - len(buf[0])]
++
++ readers = [threading.Thread(target=drain, args=(proc.stdout, "out"), daemon=True),
++ threading.Thread(target=drain, args=(proc.stderr, "err"), daemon=True)]
++ for t in readers:
++ t.start()
++ if stdin is not None:
++ try:
++ proc.stdin.write(stdin.encode("utf-8"))
++ proc.stdin.close()
++ except (BrokenPipeError, OSError):
++ pass
++ timed_out = False
++ try:
++ code = proc.wait(timeout=timeout)
++ except subprocess.TimeoutExpired:
++ timed_out = True
++ self._kill(proc)
++ code = proc.wait()
++ finally:
++ with self.mu:
++ self.procs.discard(proc)
++ for t in readers:
++ t.join(2)
++ for stream in (proc.stdout, proc.stderr, proc.stdin):
++ if stream is not None:
++ try:
++ stream.close()
++ except OSError:
++ pass
++ text = lambda b: b.decode("utf-8", "replace")
++ return {"exit_code": code, "stdout": text(got["out"][0]), "stderr": text(got["err"][0]),
++ "truncated": got["out"][1] > cap or got["err"][1] > cap, "timed_out": timed_out,
++ "duration_ms": int((self.clock() - started) * 1000)}
++
++ # ---- files -----------------------------------------------------------------------------------------------------
++ @staticmethod
++ def _kind(mode):
++ if stat.S_ISDIR(mode):
++ return "dir"
++ if stat.S_ISREG(mode):
++ return "file"
++ if stat.S_ISLNK(mode):
++ return "link"
++ return "other"
++
++ def _fs_read_op(self, op, v, level):
++ path = self._path(v.get("path"), level)
++ try:
++ st = os.stat(path)
++ except FileNotFoundError:
++ raise Fault("bad_value", "%s does not exist" % path)
++ except PermissionError:
++ raise Fault("not_allowed", "the Node's account may not read %s" % path)
++ if op == "stat":
++ return {"path": path, "type": self._kind(st.st_mode), "size": st.st_size, "mtime": int(st.st_mtime),
++ "mode": oct(stat.S_IMODE(st.st_mode))}
++ if op == "list":
++ if not stat.S_ISDIR(st.st_mode):
++ raise Fault("bad_value", "%s is not a folder" % path)
++ entries, total = [], 0
++ try:
++ with os.scandir(path) as it:
++ for e in sorted(it, key=lambda e: e.name):
++ total += 1
++ if len(entries) >= MAX_LIST:
++ continue
++ try:
++ s = e.stat(follow_symlinks=False)
++ entries.append({"name": e.name, "type": self._kind(s.st_mode), "size": s.st_size,
++ "mtime": int(s.st_mtime)})
++ except OSError:
++ entries.append({"name": e.name, "type": "other"})
++ except PermissionError:
++ raise Fault("not_allowed", "the Node's account may not list %s" % path)
++ return {"path": path, "entries": entries, "total": total, "truncated": total > len(entries)}
++ if not stat.S_ISREG(st.st_mode):
++ raise Fault("bad_value", "%s is not a file" % path)
++ offset = _int(v.get("offset"), "offset", 0, 2 ** 53, 0)
++ length = _int(v.get("length"), "length", 0, self.max_read, self.max_read)
++ try:
++ with open(path, "rb") as f:
++ f.seek(offset)
++ data = f.read(length)
++ except PermissionError:
++ raise Fault("not_allowed", "the Node's account may not read %s" % path)
++ return {"path": path, "size": st.st_size, "offset": offset, "length": len(data),
++ "truncated": offset + len(data) < st.st_size, "content_b64": base64.b64encode(data).decode("ascii")}
++
++ def _fs_write_op(self, op, v, level):
++ path = self._path(v.get("path"), level, create=True)
++ if op == "mkdir":
++ def run():
++ existed = os.path.isdir(path)
++ try:
++ os.makedirs(path, exist_ok=True)
++ except FileExistsError:
++ raise Fault("bad_value", "%s exists and is not a folder" % path)
++ except PermissionError:
++ raise Fault("not_allowed", "the Node's account may not create %s" % path)
++ return {"path": path, "created": not existed}
++ return run, "make the folder %s" % path
++ mode = v.get("mode") or "create"
++ if mode not in ("create", "overwrite", "append"):
++ raise Fault("bad_value", "mode is create, overwrite or append")
++ raw = v.get("content_b64")
++ if not isinstance(raw, str):
++ raise Fault("bad_value", "content_b64 is base64 text")
++ try:
++ data = base64.b64decode(raw, validate=True)
++ except (ValueError, TypeError):
++ raise Fault("bad_value", "content_b64 is not base64")
++ if len(data) > self.max_read:
++ raise Fault("bad_value", "at most %d bytes per write" % self.max_read)
++
++ def run():
++ if os.path.isdir(path):
++ raise Fault("bad_value", "%s is a folder" % path)
++ if mode == "create" and os.path.lexists(path):
++ raise Fault("bad_value", "%s exists (mode overwrite or append replaces or extends it)" % path)
++ try:
++ if mode == "append":
++ with open(path, "ab") as f:
++ f.write(data)
++ else:
++ fd, tmp = tempfile.mkstemp(prefix=".openvibe-", dir=os.path.dirname(path))
++ try:
++ with os.fdopen(fd, "wb") as f:
++ f.write(data)
++ if os.path.exists(path):
++ shutil.copymode(path, tmp)
++ os.replace(tmp, path)
++ except BaseException:
++ if os.path.exists(tmp):
++ os.unlink(tmp)
++ raise
++ except PermissionError:
++ raise Fault("not_allowed", "the Node's account may not write %s" % path)
++ return {"path": path, "written": len(data), "mode": mode}
++
++ verb = {"create": "create", "overwrite": "overwrite", "append": "append to"}[mode]
++ return run, "%s %s (%d bytes)" % (verb, path, len(data))
++
++ # ---- screen ----------------------------------------------------------------------------------------------------
++ def _screen_tool(self):
++ """(name, argv builder) for this machine's screen capture, or None."""
++ if self.platform == "darwin" and self.which("screencapture"):
++ return ("screencapture", lambda out: ["screencapture", "-x", "-t", "png", out])
++ if self.platform.startswith("linux"):
++ if self.env.get("WAYLAND_DISPLAY") and self.which("grim"):
++ return ("grim", lambda out: ["grim", out])
++ if self.env.get("DISPLAY"):
++ if self.which("import"):
++ return ("import", lambda out: ["import", "-window", "root", out])
++ if self.which("scrot"):
++ return ("scrot", lambda out: ["scrot", "-o", out])
++ if self.which("gnome-screenshot"):
++ return ("gnome-screenshot", lambda out: ["gnome-screenshot", "-f", out])
++ return None
++
++ def _screen(self, tool, max_width):
++ fd, out = tempfile.mkstemp(prefix="openvibe-screen-", suffix=".png")
++ os.close(fd)
++ try:
++ proc = self.popen(tool[1](out), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
++ stderr=subprocess.PIPE, start_new_session=True)
++ try:
++ _, err = proc.communicate(timeout=15)
++ except subprocess.TimeoutExpired:
++ self._kill(proc)
++ raise Fault("hardware", "the screen capture did not finish in 15 s")
++ if proc.returncode != 0 or not os.path.getsize(out):
++ raise Fault("hardware", "screen capture failed: %s" % (err or b"").decode("utf-8", "replace")[:200])
++ return self._encode(out, max_width)
++ finally:
++ if os.path.exists(out):
++ os.unlink(out)
++
++ @staticmethod
++ def _encode(path, max_width):
++ try:
++ from PIL import Image
++ except ImportError:
++ with open(path, "rb") as f:
++ png = f.read()
++ if len(png) > SCREEN_MAX_BYTES:
++ raise Fault("unsupported", "install Pillow (pip install pillow) to send screens larger than 300 KB")
++ return {"format": "png", "content_b64": base64.b64encode(png).decode("ascii")}
++ import io
++ with Image.open(path) as im:
++ im = im.convert("RGB")
++ width = min(max_width, im.width)
++ quality = 75
++ while True:
++ height = max(1, round(im.height * width / im.width))
++ frame = im.resize((width, height)) if width != im.width else im
++ buf = io.BytesIO()
++ frame.save(buf, "JPEG", quality=quality, optimize=True)
++ if buf.tell() <= SCREEN_MAX_BYTES or width <= 320:
++ break
++ if quality > 50:
++ quality -= 15
++ else:
++ width = max(320, int(width * 0.75))
++ return {"format": "jpeg", "width": width, "height": height,
++ "content_b64": base64.b64encode(buf.getvalue()).decode("ascii")}
ddiff --git a/plugins/computer/openvibe_computer/__main__.py b/plugins/computer/openvibe_computer/__main__.py
nnew file mode 100644
iindex 0000000..bfd40b6
---- /dev/null
++++ b/plugins/computer/openvibe_computer/__main__.py
@@@ -0,0 +1,6 @@
++import sys
++
++from openvibe_computer import Computer
++from openvibe_plugin import run
++
++sys.exit(run(Computer()))
ddiff --git a/plugins/computer/pyproject.toml b/plugins/computer/pyproject.toml
nnew file mode 100644
iindex 0000000..a03d3cc
---- /dev/null
++++ b/plugins/computer/pyproject.toml
@@@ -0,0 +1,17 @@
++[build-system]
++requires = ["setuptools>=61"]
++build-backend = "setuptools.build_meta"
++
++[project]
++name = "openvibe-computer"
++version = "0.1.0"
++description = "OpenVibe Node computer plugin: the machine's terminal, files and screen under the owner's control level"
++requires-python = ">=3.9"
++license = { text = "AGPL-3.0-or-later" }
++dependencies = ["openvibe-plugin"]
++
++[project.optional-dependencies]
++screen = ["pillow"]
++
++[tool.setuptools]
++packages = ["openvibe_computer"]
ddiff --git a/plugins/computer/tests/test_computer.py b/plugins/computer/tests/test_computer.py
nnew file mode 100644
iindex 0000000..f30cc1b
---- /dev/null
++++ b/plugins/computer/tests/test_computer.py
@@@ -0,0 +1,234 @@
++import base64
++import io
++import json
++import os
++import sys
++import tempfile
++import threading
++import time
++import unittest
++
++HERE = os.path.dirname(os.path.abspath(__file__))
++sys.path.insert(0, os.path.dirname(HERE))
++sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(HERE)), "sdk"))
++
++from openvibe_computer import Computer # noqa: E402
++from openvibe_plugin import Runtime # noqa: E402
++
++
++class Out(io.StringIO):
++ """The plugin's stdout, readable while worker threads write to it."""
++
++ def __init__(self):
++ super().__init__()
++ self.cond = threading.Condition()
++
++ def write(self, s):
++ with self.cond:
++ n = super().write(s)
++ self.cond.notify_all()
++ return n
++
++ def msgs(self):
++ with self.cond:
++ return [json.loads(x) for x in self.getvalue().splitlines() if x]
++
++ def wait_for(self, pred, timeout=10):
++ end = time.monotonic() + timeout
++ with self.cond:
++ while True:
++ found = [m for m in (json.loads(x) for x in self.getvalue().splitlines() if x) if pred(m)]
++ if found:
++ return found[0]
++ left = end - time.monotonic()
++ if left <= 0:
++ raise AssertionError("no matching message; got %r" % self.getvalue()[-2000:])
++ self.cond.wait(left)
++
++
++def reply(out, cid):
++ return out.wait_for(lambda m: m.get("id") == cid and m.get("op") in ("ack", "nack"))
++
++
++class ComputerTest(unittest.TestCase):
++ def boot(self, level="observe", **config):
++ self.dir = tempfile.mkdtemp(prefix="ov-computer-")
++ self.addCleanup(lambda: __import__("shutil").rmtree(self.dir, ignore_errors=True))
++ with open(os.path.join(self.dir, "notes.txt"), "w") as f:
++ f.write("hello computer\n")
++ os.mkdir(os.path.join(self.dir, "sub"))
++ self.out = Out()
++ self.plugin = Computer(env={}, platform="linux")
++ self.rt = Runtime(self.plugin, infile=io.StringIO(""), outfile=self.out)
++ cfg = {"level": level, "roots": [self.dir], **config}
++ self.rt.dispatch({"op": "hello", "config": cfg})
++ self.n = 0
++ return self.out.msgs()
++
++ def cmd(self, kind, value, deadline_ms=30000):
++ self.n += 1
++ cid = "c%d" % self.n
++ self.rt.dispatch({"op": "command", "id": cid, "kind": kind, "value": value, "deadline_ms": deadline_ms})
++ return cid
++
++ def test_describe_and_ready(self):
++ msgs = self.boot("trusted")
++ describe = msgs[0]
++ self.assertEqual(describe["driver"], "computer")
++ caps = describe["capabilities"]
++ self.assertEqual(caps["files"]["roots"], [os.path.realpath(self.dir)])
++ self.assertEqual(caps["terminal"]["level"], "trusted")
++ self.assertNotIn("screen", caps, "no display in this environment: no screen capability")
++ self.assertEqual(msgs[-1]["op"], "ready")
++
++ def test_bad_level_is_a_fault(self):
++ out = Out()
++ rt = Runtime(Computer(env={}), infile=io.StringIO(""), outfile=out)
++ rt.dispatch({"op": "hello", "config": {"level": "god"}})
++ self.assertEqual(out.msgs()[-1]["op"], "fault")
++
++ def test_observe_reads_inside_roots_only(self):
++ self.boot("observe")
++ r = reply(self.out, self.cmd("fs", {"op": "list", "path": self.dir}))
++ self.assertEqual(r["op"], "ack")
++ self.assertEqual([e["name"] for e in r["data"]["entries"]], ["notes.txt", "sub"])
++ r = reply(self.out, self.cmd("fs", {"op": "read", "path": os.path.join(self.dir, "notes.txt")}))
++ self.assertEqual(base64.b64decode(r["data"]["content_b64"]), b"hello computer\n")
++ r = reply(self.out, self.cmd("fs", {"op": "read", "path": os.path.join(self.dir, "notes.txt"), "offset": 6, "length": 3}))
++ self.assertEqual((base64.b64decode(r["data"]["content_b64"]), r["data"]["truncated"]), (b"com", True))
++ r = reply(self.out, self.cmd("fs", {"op": "stat", "path": "/etc/hostname"}))
++ self.assertEqual((r["op"], r["fault_code"]), ("nack", "not_allowed"), "outside the roots")
++ link = os.path.join(self.dir, "escape")
++ os.symlink("/etc", link)
++ r = reply(self.out, self.cmd("fs", {"op": "list", "path": link}))
++ self.assertEqual(r["fault_code"], "not_allowed", "a symlink out of the roots is outside them")
++ r = reply(self.out, self.cmd("fs", {"op": "list", "path": "relative/path"}))
++ self.assertEqual(r["fault_code"], "bad_value")
++
++ def test_observe_refuses_acting(self):
++ self.boot("observe")
++ r = reply(self.out, self.cmd("exec", {"argv": ["echo", "hi"]}))
++ self.assertEqual((r["op"], r["fault_code"]), ("nack", "not_allowed"))
++ r = reply(self.out, self.cmd("fs", {"op": "write", "path": os.path.join(self.dir, "x"), "content_b64": "eA=="}))
++ self.assertEqual(r["fault_code"], "not_allowed")
++ self.assertFalse(os.path.exists(os.path.join(self.dir, "x")))
++
++ def test_trusted_runs_programs_and_writes(self):
++ self.boot("trusted")
++ r = reply(self.out, self.cmd("exec", {"argv": [sys.executable, "-c", "import os,sys; print(os.getcwd()); sys.stderr.write('e'); sys.exit(3)"]}))
++ self.assertEqual(r["op"], "ack")
++ d = r["data"]
++ self.assertEqual((d["exit_code"], d["stdout"].strip(), d["stderr"], d["timed_out"]), (3, os.path.realpath(self.dir), "e", False))
++ r = reply(self.out, self.cmd("exec", {"argv": ["cat"], "stdin": "piped"}))
++ self.assertEqual(r["data"]["stdout"], "piped")
++ r = reply(self.out, self.cmd("exec", {"argv": [sys.executable, "-c", "print('x' * 5000)"]}))
++ self.assertFalse(r["data"]["truncated"])
++ r = reply(self.out, self.cmd("exec", {"argv": ["definitely-not-a-program-ov"]}))
++ self.assertEqual(r["fault_code"], "bad_value")
++ r = reply(self.out, self.cmd("exec", {"argv": "rm -rf /"}))
++ self.assertEqual(r["fault_code"], "bad_value", "a shell line is not argv")
++ target = os.path.join(self.dir, "sub", "new.txt")
++ r = reply(self.out, self.cmd("fs", {"op": "write", "path": target, "content_b64": base64.b64encode(b"one").decode()}))
++ self.assertEqual((r["op"], r["data"]["written"]), ("ack", 3))
++ r = reply(self.out, self.cmd("fs", {"op": "write", "path": target, "content_b64": "dHdv"}))
++ self.assertEqual(r["fault_code"], "bad_value", "create does not replace")
++ reply(self.out, self.cmd("fs", {"op": "write", "path": target, "content_b64": "dHdv", "mode": "append"}))
++ with open(target, "rb") as f:
++ self.assertEqual(f.read(), b"onetwo")
++ r = reply(self.out, self.cmd("fs", {"op": "mkdir", "path": os.path.join(self.dir, "a", "b")}))
++ self.assertEqual(r["data"]["created"], True)
++ r = reply(self.out, self.cmd("fs", {"op": "write", "path": "/tmp/ov-outside-roots", "content_b64": "eA=="}))
++ self.assertEqual(r["fault_code"], "not_allowed", "trusted stays inside the roots")
++
++ def test_output_is_capped_and_timeouts_kill(self):
++ self.boot("trusted", max_output_bytes=1024, timeout_s=1)
++ r = reply(self.out, self.cmd("exec", {"argv": [sys.executable, "-c", "print('y' * 100000)"]}))
++ self.assertEqual((len(r["data"]["stdout"]), r["data"]["truncated"]), (1024, True))
++ r = reply(self.out, self.cmd("exec", {"argv": [sys.executable, "-c", "import time; time.sleep(30)"], "timeout_s": 60}))
++ self.assertEqual((r["data"]["timed_out"], r["data"]["exit_code"] != 0), (True, True))
++ self.assertLess(r["data"]["duration_ms"], 5000, "the owner's timeout_s caps the request's")
++
++ def test_ask_holds_until_the_owner_answers(self):
++ self.boot("ask")
++ r = reply(self.out, self.cmd("fs", {"op": "list", "path": self.dir}))
++ self.assertEqual(r["op"], "ack", "observing never asks")
++ cid = self.cmd("exec", {"argv": ["echo", "approved"]})
++ ev = self.out.wait_for(lambda m: m.get("op") == "event" and m.get("name") == "approval")
++ self.assertEqual((ev["id"], ev["kind"]), (cid, "exec"))
++ self.assertIn("echo approved", ev["summary"])
++ self.assertFalse([m for m in self.out.msgs() if m.get("id") == cid and m.get("op") in ("ack", "nack")])
++ self.rt.dispatch({"op": "approval", "id": cid, "approved": True})
++ self.assertEqual(reply(self.out, cid)["data"]["stdout"], "approved\n")
++ cid = self.cmd("fs", {"op": "write", "path": os.path.join(self.dir, "denied.txt"), "content_b64": "eA=="})
++ self.out.wait_for(lambda m: m.get("name") == "approval" and m.get("id") == cid)
++ self.rt.dispatch({"op": "approval", "id": cid, "approved": False})
++ r = reply(self.out, cid)
++ self.assertEqual((r["fault_code"], r["message"]), ("not_allowed", "the owner denied it on the machine"))
++ self.assertFalse(os.path.exists(os.path.join(self.dir, "denied.txt")))
++
++ def test_unanswered_requests_expire(self):
++ self.boot("ask", approval_timeout_s=10)
++ t = [1000.0]
++ self.plugin.clock = lambda: t[0]
++ cid = self.cmd("exec", {"argv": ["echo", "late"]})
++ self.plugin.pending[cid] = (self.plugin.pending[cid][0], t[0] + 10, "x")
++ t[0] += 11
++ tele = self.plugin.poll()
++ self.assertEqual((tele["sensors"]["computer_waiting"], tele["sensors"]["computer_level"]), (0, "ask"))
++ r = reply(self.out, cid)
++ self.assertEqual(r["fault_code"], "not_allowed")
++
++ def test_the_kill_switch_drops_to_observe_and_denies_what_waits(self):
++ self.boot("ask")
++ cid = self.cmd("exec", {"argv": ["echo", "never"]})
++ self.out.wait_for(lambda m: m.get("name") == "approval")
++ self.rt.dispatch({"op": "estop"})
++ self.assertEqual(reply(self.out, cid)["fault_code"], "local_stop")
++ self.assertEqual(self.plugin.effective_level(), "observe")
++ r = reply(self.out, self.cmd("exec", {"argv": ["echo", "no"]}))
++ self.assertEqual(r["fault_code"], "local_stop")
++ self.assertEqual(reply(self.out, self.cmd("fs", {"op": "stat", "path": self.dir}))["op"], "ack", "observing still works")
++ self.rt.dispatch({"op": "resume"})
++ self.assertEqual(self.plugin.effective_level(), "ask")
++
++ def test_halt_kills_running_programs(self):
++ self.boot("trusted", timeout_s=60)
++ cid = self.cmd("exec", {"argv": [sys.executable, "-c", "import time; time.sleep(60)"]})
++ end = time.monotonic() + 5
++ while not self.plugin.procs and time.monotonic() < end:
++ time.sleep(0.02)
++ self.rt.dispatch({"op": "command", "id": "h", "kind": "halt"})
++ r = reply(self.out, cid)
++ self.assertNotEqual(r["data"]["exit_code"], 0)
++
++ def test_full_reaches_outside_the_roots(self):
++ self.boot("full")
++ r = reply(self.out, self.cmd("fs", {"op": "stat", "path": "/"}))
++ self.assertEqual((r["op"], r["data"]["type"]), ("ack", "dir"))
++
++ def test_screen_tool_detection(self):
++ which = lambda name: "/usr/bin/" + name if name in ("grim", "scrot") else None
++ self.assertEqual(Computer(which=which, env={"WAYLAND_DISPLAY": "wayland-0"}, platform="linux")._screen_tool()[0], "grim")
++ self.assertEqual(Computer(which=which, env={"DISPLAY": ":0"}, platform="linux")._screen_tool()[0], "scrot")
++ self.assertIsNone(Computer(which=which, env={}, platform="linux")._screen_tool())
++ mac = Computer(which=lambda n: "/usr/sbin/screencapture" if n == "screencapture" else None, env={}, platform="darwin")
++ self.assertEqual(mac._screen_tool()[1]("/tmp/x.png"), ["screencapture", "-x", "-t", "png", "/tmp/x.png"])
++
++ def test_screen_capture_encodes_a_jpeg(self):
++ try:
++ from PIL import Image
++ except ImportError:
++ self.skipTest("pillow not installed")
++ self.boot("observe")
++ src = os.path.join(self.dir, "frame.png")
++ Image.new("RGB", (2560, 1440), (30, 90, 200)).save(src)
++ tool = ("fake", lambda out: [sys.executable, "-c", "import shutil,sys; shutil.copy(sys.argv[1], sys.argv[2])", src, out])
++ self.plugin._screen_tool = lambda: tool
++ r = reply(self.out, self.cmd("screen", {"max_width": 1280}))
++ d = r["data"]
++ self.assertEqual((d["format"], d["width"], d["height"]), ("jpeg", 1280, 720))
++ self.assertEqual(base64.b64decode(d["content_b64"])[:2], b"\xff\xd8")
++
++
++if __name__ == "__main__":
++ unittest.main()
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
iindex 68a600c..962ba03 100644
---- a/plugins/sdk/openvibe_plugin/__init__.py
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -19,13 +19,15 @@ import sys
import threading
import time
--__all__ = ["Plugin", "Fault", "Command", "Runtime", "run", "clamp", "MOTION_KINDS", "HEARTBEAT_TIMEOUT_S",
++__all__ = ["Plugin", "Fault", "Deferred", "Command", "Runtime", "run", "clamp", "MOTION_KINDS", "HEARTBEAT_TIMEOUT_S",
"DEFAULT_DEADLINE_MS", "PROTOCOL_VERSION"]
PROTOCOL_VERSION = 1
HEARTBEAT_TIMEOUT_S = 1.0
DEFAULT_DEADLINE_MS = 300
MAX_DEADLINE_MS = 5000
++# Kinds that move nothing (a computer's exec, fs and screen) keep a long deadline: the plugin answers within it.
++MAX_LONG_DEADLINE_MS = 600000
# Kinds that move something continuously and therefore stop at their deadline. A plugin may widen this set
# (Plugin.motion_kinds) for actuators it drives by speed rather than by position.
MOTION_KINDS = frozenset({"drive"})
@@@ -42,6 +44,19 @@ class Fault(Exception):
self.message = message
++class Deferred(Exception):
++ """Raised by handle() to answer later, with Plugin.complete(cmd.id, data) or Plugin.fail(cmd.id, code, message):
++ slow work done on a thread of the plugin's own, or a command held for its owner's approval on the machine. The core
++ waits for the answer until the command's deadline_ms (computer kinds carry long ones)."""
++
++
++def _ack(cid, data=None):
++ out = {"op": "ack", "id": cid}
++ if data is not None:
++ out["data"] = data
++ return out
++
++
class Command:
__slots__ = ("id", "kind", "value", "deadline_ms", "deadline", "target")
@@@ -93,7 +108,8 @@ class Plugin:
"""Open the hardware. Every actuator must be stopped when this returns."""
def handle(self, cmd):
-- """Execute a command. Return normally for ack, raise Fault for nack."""
++ """Execute a command. Return normally for ack (a dict returned is the ack's data: what a computer command
++ answers), raise Fault for nack, raise Deferred to answer later with complete() or fail()."""
raise Fault("unsupported", "kind %s not supported" % cmd.kind)
def stop_motion(self):
@@@ -110,7 +126,24 @@ class Plugin:
def close(self):
"""Release the hardware. stop() has already been called."""
++ def on_approval(self, cid, approved):
++ """The owner answered, on the machine (`openvibe-node approve|deny <id>`), an approval this plugin asked for
++ with emit_event("approval", id=…). Runs on the runtime's main thread."""
++
# Helpers for drivers.
++ def complete(self, cid, data=None):
++ """Answer a deferred command with ack (and its data). Safe from any thread."""
++ if self.runtime is not None:
++ self.runtime.send(_ack(cid, data))
++
++ def fail(self, cid, code, message=""):
++ """Answer a deferred command with nack. Safe from any thread."""
++ if self.runtime is not None:
++ out = {"op": "nack", "id": cid, "fault_code": code}
++ if message:
++ out["message"] = str(message)[:500]
++ self.runtime.send(out)
++
def emit_event(self, name, **fields):
if self.runtime is not None:
self.runtime.send(dict(op="event", name=name, **fields))
@@@ -237,6 +270,12 @@ class Runtime:
self.estopped = True
self.local_stop_pending = False
self.safe_stop("estop")
++ elif op == "approval":
++ try:
++ with self.lock:
++ self.plugin.on_approval(msg.get("id"), msg.get("approved") is True)
++ except Exception as e:
++ self.plugin.log("approval handler failed: %r", e)
elif op == "resume":
if self.local_stop_pending:
self.plugin.log("resume ignored: the local stop has not reached the core yet")
@@@ -288,7 +327,8 @@ class Runtime:
dl = int(msg.get("deadline_ms") or DEFAULT_DEADLINE_MS)
except (TypeError, ValueError):
dl = DEFAULT_DEADLINE_MS
-- dl = max(1, min(dl, MAX_DEADLINE_MS))
++ moves = kind in GUARDED_KINDS or kind in self.plugin.motion_kinds
++ dl = max(1, min(dl, MAX_DEADLINE_MS if moves else MAX_LONG_DEADLINE_MS))
cmd = Command(cid, kind, msg.get("value"), dl, now, msg.get("target"))
if kind == "halt":
self.safe_stop("halt")
@@@ -306,10 +346,12 @@ class Runtime:
return
try:
with self.lock:
-- self.plugin.handle(cmd)
++ data = self.plugin.handle(cmd)
if kind in self.plugin.motion_kinds:
self.motion_deadline = cmd.deadline
-- self.send({"op": "ack", "id": cid})
++ self.send(_ack(cid, data))
++ except Deferred:
++ return
except Fault as f:
out = {"op": "nack", "id": cid, "fault_code": f.code}
if f.message:
ddiff --git a/plugins/sdk/tests/test_runtime.py b/plugins/sdk/tests/test_runtime.py
iindex 54cd6d3..f7f03fd 100644
---- a/plugins/sdk/tests/test_runtime.py
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -9,7 +9,7 @@ import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(HERE))
--from openvibe_plugin import Fault, Plugin, Runtime # noqa: E402
++from openvibe_plugin import Deferred, Fault, Plugin, Runtime # noqa: E402
class FakeClock:
@@@ -260,5 +260,55 @@ class ProcessTest(unittest.TestCase):
self.assertIn("C noise", err)
++class Later(Plugin):
++ """Answers `ask` later (with data) and `look` at once with data; remembers approvals."""
++ driver = "later"
++ motion_kinds = frozenset()
++
++ def __init__(self):
++ self.held = []
++ self.answers = []
++
++ def handle(self, cmd):
++ if cmd.kind == "look":
++ return {"seen": cmd.value.get("what"), "deadline_ms": cmd.deadline_ms}
++ if cmd.kind == "ask":
++ self.held.append(cmd.id)
++ raise Deferred()
++ raise Fault("unsupported")
++
++ def on_approval(self, cid, approved):
++ self.answers.append((cid, approved))
++ if approved:
++ self.complete(cid, {"done": True})
++ else:
++ self.fail(cid, "not_allowed", "denied")
++
++
++class DeferredTest(unittest.TestCase):
++ def test_data_deferred_and_approval(self):
++ rt, p, out, clock = make(Later())
++ rt.dispatch({"op": "hello", "config": {}})
++ rt.dispatch({"op": "command", "id": "a", "kind": "look", "value": {"what": "screen"}, "deadline_ms": 60000})
++ ack = lines(out)[-1]
++ self.assertEqual(ack, {"op": "ack", "id": "a", "data": {"seen": "screen", "deadline_ms": 60000}},
++ "a kind that moves nothing keeps its long deadline")
++ rt.dispatch({"op": "command", "id": "b", "kind": "ask", "value": {}})
++ self.assertEqual(p.held, ["b"])
++ self.assertFalse([m for m in lines(out) if m.get("id") == "b"], "no answer until the plugin gives one")
++ rt.dispatch({"op": "approval", "id": "b", "approved": True})
++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "b", "data": {"done": True}})
++ rt.dispatch({"op": "command", "id": "c", "kind": "ask", "value": {}})
++ rt.dispatch({"op": "approval", "id": "c", "approved": "yes"})
++ self.assertEqual(p.answers[-1], ("c", False), "only a JSON true approves")
++ self.assertEqual(lines(out)[-1], {"op": "nack", "id": "c", "fault_code": "not_allowed", "message": "denied"})
++
++ def test_motion_kinds_keep_the_short_cap(self):
++ rt, p, out, clock = make()
++ rt.dispatch({"op": "hello", "config": {}})
++ rt.dispatch({"op": "command", "id": "d", "kind": "drive", "value": {}, "deadline_ms": 60000})
++ self.assertEqual(rt.motion_deadline, clock() + 5.0)
++
++
if __name__ == "__main__":
unittest.main()
ddiff --git a/scripts/dist.sh b/scripts/dist.sh
iindex 1d2b7bd..a8fa1a4 100755
---- a/scripts/dist.sh
++++ b/scripts/dist.sh
@@@ -24,7 +24,7 @@ done
# The plugin bundle: the Python packages without their tests.
tar -czf dist/openvibe-node-plugins.tar.gz -C plugins \
--exclude=tests --exclude=__pycache__ --exclude='*.egg-info' --exclude=.pytest_cache --exclude=build \
-- sdk dryrun adeept_adr036 cozmo relay ros2
++ sdk dryrun adeept_adr036 cozmo relay ros2 computer
# OpenVibe.Bot's GET /install route and BOT_INSTALLER_SOURCE_URL are documented to serve this asset name:
# https://github.com/OpenVibers/OpenVibe.Node/releases/latest/download/install.sh — do not rename it.
cp install/install.sh dist/install.sh
ddiff --git a/scripts/test-install-computer-config.sh b/scripts/test-install-computer-config.sh
nnew file mode 100644
iindex 0000000..93fe491
---- /dev/null
++++ b/scripts/test-install-computer-config.sh
@@@ -0,0 +1,26 @@
++#!/bin/sh
++# Tests install.sh --computer's config: OPENVIBE_INSTALL_COMPUTER_CONFIG_ONLY=<file> adds the computer plugin at the
++# level given, its folder the owner's home, in place of the dry-run plugin and of an earlier computer entry; the rest of
++# the config stays, and a level that is not one of the four is refused.
++set -eu
++
++here="$(cd "$(dirname "$0")" && pwd)"
++installer="$here/../install/install.sh"
++tmp="$(mktemp -d)"
++trap 'rm -rf "$tmp"' EXIT
++fail() { printf 'FAIL: %s\n' "$*" >&2; exit 1; }
++
++conf="$tmp/config.json"
++printf '{"server": "https://openvibe.bot", "plugins": [{"name": "dryrun"}, {"name": "ros2"}]}\n' > "$conf"
++OPENVIBE_INSTALL_COMPUTER_CONFIG_ONLY="$conf" OPENVIBE_COMPUTER_HOME=/home/someone sh "$installer" --computer ask
++OPENVIBE_INSTALL_COMPUTER_CONFIG_ONLY="$conf" OPENVIBE_COMPUTER_HOME=/home/someone sh "$installer" --computer trusted
++python3 - "$conf" <<'PY' || fail "config"
++import json, sys
++c = json.load(open(sys.argv[1]))
++assert c['server'] == 'https://openvibe.bot', 'the rest of the config stays'
++assert c['plugins'] == [{'name': 'ros2'}, {'name': 'computer', 'config': {'level': 'trusted', 'roots': ['/home/someone']}}], c['plugins']
++PY
++if OPENVIBE_INSTALL_COMPUTER_CONFIG_ONLY="$conf" OPENVIBE_COMPUTER_HOME=/home/someone sh "$installer" --computer root 2>/dev/null; then
++ fail "an unknown level was accepted"
++fi
++echo "ok: install.sh --computer config"