openvibers/openvibe.link Commits

Node: probe — a read-only check of the machine's own network (Watch's node carrier) (#43)

OP OpenVibers committed 30 days ago · fc816114e5 · 7 files, +159 −3

Plan T18 step 7 (Contracts 0.167.0): the Node side of Watch's `node` carrier. The computer plugin (0.3.0) gains `probe`, a read-only check of something on the machine's own network.

- **`http(s)://…`**: one GET, with no redirect followed (a 3xx is reported with its `Location`). The answer: `status`, `content_type`, `etag`, `last_modified`, `body` (at most 64 KB) and `ms`.
- **`tcp://host:port`**: `open` true or false, and `ms`.
- **Own network only**: every address the name resolves to must be private, loopback or link-local, and the connection goes to the address that was checked, so a second DNS answer cannot redirect it. Public targets are refused with `not_allowed`; credentials in the URL are refused.
- **At every level**: it is read-only, so it runs at every level, including while the local stop is latched (the core treats it like reading).
- **Config**: `probe: false` turns it off.
- **Core**: kind `probe`, capability `probe` in describe.
- **Docs**: `docs/computer.md` and `protocol.md`.

Tests: two plugin tests (a loopback HTTP server with ETag and a redirect that is not followed; TCP open and closed; public IPv4 and IPv6 refused; bad targets; runs while stopped; off) and an end-to-end check through the core. `go test -race ./...` passes, as do the plugin tests (16).

Changes

ddiff --git a/docs/computer.md b/docs/computer.md
iindex a220df1..9b33299 100644
---- a/docs/computer.md
++++ b/docs/computer.md
@@@ -56,6 +56,7 @@ waits is denied, and the core also refuses `exec` and file writes until `openvib
|----------|-----------------------------------------------------------------------------------------------|---------------------|
| `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` |
++| `probe` | `target`: `http(s)://…` (one GET, no redirect followed) or `tcp://host:port`; `timeout_s` | `status`, `content_type`, `etag`, `last_modified`, `body` (≤ 64 KB), `ms`; or `open`, `ms` |
| `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
@@@ -75,6 +76,7 @@ your graphical session for `screen`; without one the plugin simply does not offe
| `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 |
| `notify` | `true` | a desktop notification when a request waits (graphical session only) |
++| `probe` | `true` | `false`: no probes (OpenVibe.Watch's `node` source) |
`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
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 1e78324..ca8a8a6 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -201,6 +201,7 @@ at most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message
| `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 |
++| `probe` | `{"target": "http(s)://…" \| "tcp://host:port", "timeout_s"?}`: a read-only check of the machine's own network | 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
ddiff --git a/internal/node/computer_test.go b/internal/node/computer_test.go
iindex d1c9390..bbd5d20 100644
---- a/internal/node/computer_test.go
++++ b/internal/node/computer_test.go
@@@ -221,6 +221,12 @@ func TestComputerLocalStopLeavesObserveOnly(t *testing.T) {
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 d := dataOf(t, e.computerCmd("s3p", protocol.KindProbe, map[string]any{"target": "tcp://127.0.0.1:1", "timeout_s": 2}, 10*time.Second)); d["open"] != false {
++ t.Fatalf("a probe is read-only and runs while stopped: %v", d)
++ }
++ if nk := nackOf(t, e.computerCmd("s3q", protocol.KindProbe, map[string]any{"target": "http://1.1.1.1/"}, 10*time.Second)); nk.FaultCode != protocol.FaultNotAllowed {
++ t.Fatalf("a public probe: %+v", nk)
++ }
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)
}
ddiff --git a/internal/plugins/manager.go b/internal/plugins/manager.go
iindex be2d7a0..c06400b 100644
---- a/internal/plugins/manager.go
++++ b/internal/plugins/manager.go
@@@ -20,6 +20,7 @@ var capabilityFor = map[string]string{
protocol.KindExec: "terminal",
protocol.KindFS: "files",
protocol.KindScreen: "screen",
++ protocol.KindProbe: "probe",
}
// Manager supervises every configured plugin.
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 9d24877..ddbf47e 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -161,13 +161,16 @@ const (
KindExec = "exec"
KindFS = "fs"
KindScreen = "screen"
++ // KindProbe is a read-only check of something on the machine's own network (one GET or a TCP connect to a private,
++ // loopback or link-local address): OpenVibe.Watch's node carrier.
++ KindProbe = "probe"
)
// Kinds lists every command kind the Node understands.
--var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt, KindButton, KindPoint, KindExec, KindFS, KindScreen}
++var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt, KindButton, KindPoint, KindExec, KindFS, KindScreen, KindProbe}
// 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}
++var ComputerKinds = map[string]bool{KindExec: true, KindFS: true, KindScreen: true, KindProbe: true}
const (
// DefaultComputerMS is a computer command's reply window when the server sends no deadline; MaxComputerMS caps
ddiff --git a/plugins/computer/openvibe_computer/__init__.py b/plugins/computer/openvibe_computer/__init__.py
iindex d35c740..7cdcec7 100644
---- a/plugins/computer/openvibe_computer/__init__.py
++++ b/plugins/computer/openvibe_computer/__init__.py
@@@ -19,10 +19,18 @@ Command kinds and their answers (the ack's ``data``):
* ``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).
++* ``probe`` ``{"target", "timeout_s"?}`` → for ``http(s)://…`` one GET, no redirect followed: ``{status, content_type,
++ etag, last_modified, body, truncated, ms}``; for ``tcp://host:port``: ``{open, ms, error?}``. Read-only, so it runs at
++ every level, and only towards the machine's own network: every address the name resolves to must be private,
++ loopback or link-local, and the connection goes to the address that was checked. Config ``probe: false`` turns it off.
"""
import base64
++import http.client
++import ipaddress
import os
++import socket
++import ssl
import shlex
import shutil
import stat
@@@ -31,6 +39,7 @@ import sys
import tempfile
import threading
import time
++import urllib.parse
from openvibe_plugin import Deferred, Fault, Plugin
@@@ -60,7 +69,7 @@ def _int(v, name, lo, hi, default):
class Computer(Plugin):
driver = "computer"
-- version = "0.2.0"
++ version = "0.3.0"
motion_kinds = frozenset()
telemetry_interval_s = 2.0
@@@ -76,6 +85,7 @@ class Computer(Plugin):
self.max_output = 64 * 1024
self.max_read = 256 * 1024
self.allow_exec = True
++ self.allow_probe = True
self.notify = True
self.approval_timeout_s = 120
self.pending = {} # command id -> (run, expires_at, summary)
@@@ -100,6 +110,7 @@ class Computer(Plugin):
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
++ self.allow_probe = config.get("probe", True) is not False
self.notify = config.get("notify", True) is not False
def describe(self, config):
@@@ -113,6 +124,8 @@ class Computer(Plugin):
tool = self._screen_tool()
if tool:
caps["screen"] = {"level": self.level, "tool": tool[0]}
++ if self.allow_probe:
++ caps["probe"] = {"lan_only": True, "schemes": ["http", "https", "tcp"]}
return caps
def setup(self, config):
@@@ -203,6 +216,11 @@ class Computer(Plugin):
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 == "probe":
++ if not self.allow_probe:
++ raise Fault("unsupported", "this machine does not allow probes")
++ target, timeout = self._probe_target(v)
++ return self._start(cmd, lambda: self._probe(target, timeout))
if cmd.kind == "exec":
if not self.allow_exec:
raise Fault("unsupported", "this machine does not allow running programs")
@@@ -376,6 +394,76 @@ class Computer(Plugin):
"truncated": got["out"][1] > cap or got["err"][1] > cap, "timed_out": timed_out,
"duration_ms": int((self.clock() - started) * 1000)}
++ # ---- probe (the node carrier of OpenVibe.Watch) -----------------------------------------------------------------
++ @staticmethod
++ def _lan(addr):
++ ip = ipaddress.ip_address(addr)
++ if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
++ ip = ip.ipv4_mapped
++ return ip.is_private or ip.is_loopback or ip.is_link_local
++
++ def _probe_target(self, v):
++ raw = v.get("target")
++ if not isinstance(raw, str) or len(raw) > 2048:
++ raise Fault("bad_value", "target is http(s)://host/path or tcp://host:port")
++ u = urllib.parse.urlsplit(raw)
++ if u.scheme not in ("http", "https", "tcp") or not u.hostname or u.username or u.password:
++ raise Fault("bad_value", "target is http(s)://host/path or tcp://host:port, with no credentials")
++ try:
++ port = u.port or {"http": 80, "https": 443}.get(u.scheme)
++ except ValueError:
++ raise Fault("bad_value", "the port is not a number")
++ if not port:
++ raise Fault("bad_value", "a tcp target names its port")
++ timeout = _int(v.get("timeout_s"), "timeout_s", 1, 30, 10)
++ return {"scheme": u.scheme, "host": u.hostname, "port": port, "path": (u.path or "/") + ("?" + u.query if u.query else ""), "raw": raw}, timeout
++
++ def _resolve_lan(self, host, port):
++ """The first address of host, when every address it resolves to is on this machine's own network."""
++ try:
++ infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
++ except socket.gaierror as e:
++ raise Fault("bad_value", "%s does not resolve: %s" % (host, e))
++ addrs = [i[4][0] for i in infos]
++ if not addrs or not all(self._lan(a) for a in addrs):
++ raise Fault("not_allowed", "a probe reaches this machine's own network only (private, loopback or link-local addresses); %s is not" % host)
++ return infos[0][0], addrs[0]
++
++ def _probe(self, t, timeout):
++ family, addr = self._resolve_lan(t["host"], t["port"])
++ started = self.clock()
++ ms = lambda: int((self.clock() - started) * 1000)
++ if t["scheme"] == "tcp":
++ try:
++ with socket.socket(family, socket.SOCK_STREAM) as s:
++ s.settimeout(timeout)
++ s.connect((addr, t["port"]))
++ return {"open": True, "ms": ms()}
++ except OSError as e:
++ return {"open": False, "ms": ms(), "error": str(e)[:200]}
++ raw = socket.create_connection((addr, t["port"]), timeout=timeout)
++ try:
++ if t["scheme"] == "https":
++ raw = ssl.create_default_context().wrap_socket(raw, server_hostname=t["host"])
++ conn = http.client.HTTPConnection(t["host"], t["port"], timeout=timeout)
++ conn.sock = raw
++ conn.request("GET", t["path"], headers={"User-Agent": "OpenVibe-Node-probe/1", "Accept": "*/*", "Connection": "close"})
++ res = conn.getresponse()
++ cap = min(self.max_output, 64 * 1024)
++ body = res.read(cap + 1)
++ return {"status": res.status, "content_type": res.getheader("Content-Type"), "etag": res.getheader("ETag"),
++ "last_modified": res.getheader("Last-Modified"), "location": res.getheader("Location"),
++ "body": body[:cap].decode("utf-8", "replace"), "truncated": len(body) > cap, "ms": ms()}
++ except ssl.SSLError as e:
++ raise Fault("hardware", "TLS failed: %s" % str(e)[:200])
++ except (OSError, http.client.HTTPException) as e:
++ raise Fault("hardware", "the probe failed: %s" % str(e)[:200])
++ finally:
++ try:
++ raw.close()
++ except OSError:
++ pass
++
# ---- files -----------------------------------------------------------------------------------------------------
@staticmethod
def _kind(mode):
ddiff --git a/plugins/computer/tests/test_computer.py b/plugins/computer/tests/test_computer.py
iindex a0fbd51..067a8c7 100644
---- a/plugins/computer/tests/test_computer.py
++++ b/plugins/computer/tests/test_computer.py
@@@ -225,6 +225,61 @@ class ComputerTest(unittest.TestCase):
r = reply(self.out, self.cmd("fs", {"op": "stat", "path": "/"}))
self.assertEqual((r["op"], r["data"]["type"]), ("ack", "dir"))
++ def test_probe_reaches_the_machines_own_network_only(self):
++ import http.server
++ import socket as socketlib
++
++ class Handler(http.server.BaseHTTPRequestHandler):
++ def do_GET(self):
++ if self.path == "/moved":
++ self.send_response(302)
++ self.send_header("Location", "http://1.1.1.1/")
++ self.end_headers()
++ return
++ body = b'{"up": true}'
++ self.send_response(200)
++ self.send_header("Content-Type", "application/json")
++ self.send_header("ETag", '"v1"')
++ self.end_headers()
++ self.wfile.write(body)
++
++ def log_message(self, *a):
++ pass
++
++ server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
++ threading.Thread(target=server.serve_forever, daemon=True).start()
++ self.addCleanup(server.shutdown)
++ port = server.server_address[1]
++ self.boot("observe")
++ r = reply(self.out, self.cmd("probe", {"target": "http://127.0.0.1:%d/status" % port}))
++ self.assertEqual(r["op"], "ack", r)
++ self.assertEqual((r["data"]["status"], r["data"]["body"], r["data"]["etag"], r["data"]["content_type"]), (200, '{"up": true}', '"v1"', "application/json"))
++ r = reply(self.out, self.cmd("probe", {"target": "http://127.0.0.1:%d/moved" % port}))
++ self.assertEqual((r["data"]["status"], r["data"]["location"]), (302, "http://1.1.1.1/"), "a redirect is reported, never followed")
++ r = reply(self.out, self.cmd("probe", {"target": "tcp://127.0.0.1:%d" % port}))
++ self.assertTrue(r["data"]["open"])
++ closed = socketlib.socket()
++ closed.bind(("127.0.0.1", 0))
++ free = closed.getsockname()[1]
++ closed.close()
++ r = reply(self.out, self.cmd("probe", {"target": "tcp://127.0.0.1:%d" % free, "timeout_s": 2}))
++ self.assertFalse(r["data"]["open"])
++ for public in ("http://1.1.1.1/", "tcp://8.8.8.8:53", "https://[2606:4700:4700::1111]/"):
++ r = reply(self.out, self.cmd("probe", {"target": public}))
++ self.assertEqual((r["op"], r["fault_code"]), ("nack", "not_allowed"), public)
++ for bad in ("ftp://127.0.0.1/", "http://user:[email protected]/", "tcp://127.0.0.1", "nonsense"):
++ r = reply(self.out, self.cmd("probe", {"target": bad}))
++ self.assertEqual(r["fault_code"], "bad_value", bad)
++ self.rt.dispatch({"op": "estop"})
++ r = reply(self.out, self.cmd("probe", {"target": "tcp://127.0.0.1:%d" % port}))
++ self.assertTrue(r["data"]["open"], "a probe is read-only: it runs while stopped")
++
++ def test_probe_can_be_turned_off(self):
++ self.boot("trusted", probe=False)
++ self.assertNotIn("probe", self.out.msgs()[0]["capabilities"])
++ r = reply(self.out, self.cmd("probe", {"target": "tcp://127.0.0.1:1"}))
++ self.assertEqual(r["fault_code"], "unsupported")
++
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")