Compare commits

...
Author SHA1 Message Date
gandalfandClaude Opus 4.7 e587a5329e docs(remote-ui/square): README + CLAUDE.md for Phase 3 (ref #127)
README rewritten for the Phase 3 architecture (Pillow+/dev/fb0 kiosk, no
Chromium/Qt/X/nginx). CLAUDE.md added with file map, stack inventory, run +
debug recipes, and the round/ untouched constraint.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:26:09 +02:00
gandalfandClaude Opus 4.7 04817781ab fix(remote-ui/square): deploy.sh targets kiosk service (drops chromium+right-panel+nginx) (ref #127)
Hot-update flow simplified for Phase 3:
- Rsync helper + kiosk Python packages only (no /var/www/, no
  square-bridge.js, no nginx content).
- Restart secubox-eye-square-helper + secubox-square-kiosk
  (was chromium + right-panel + nginx reload).
- Health check via systemctl is-active (was curl http://localhost/).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:25:28 +02:00
gandalfandClaude Opus 4.7 eb0f5e66e3 fix(remote-ui/square): firstboot.sh enables kiosk service (drops nginx + 3 Phase 2 units) (ref #127)
Phase 3 firstboot only enables secubox-otg-gadget, secubox-eye-square-helper,
and secubox-square-kiosk. The X / Chromium / right-panel / nginx units no
longer exist on the image.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:24:54 +02:00
gandalfandClaude Opus 4.7 8f71ff4ed5 fix(remote-ui/square): build-eye-square-image.sh slimmed for Phase 3 (ref #127)
Apt install list trimmed: dropped chromium, openbox, xserver-xorg, xinit,
unclutter, 12 libxcb-* libraries, nginx-light, python3-pip; added
python3-pil + python3-evdev. The PySide6 pip install is gone.

Removed Phase 2 vestiges:
- nginx + round/ html copy + square-bridge.js injection
- openbox autostart, xinitrc chmod
- render group (no /dev/dri access needed without Chromium)

Systemd enable list updated:
- Dropped: nginx, secubox-kiosk-x, secubox-square-chromium,
           secubox-square-right-panel
- Added:   secubox-square-kiosk
- Default target now multi-user.target (was graphical.target)

Python install copies kiosk package (was right_panel package).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:23:06 +02:00
gandalfandClaude Opus 4.7 f165429f85 fix(secubox-eye-square): debian/control drops Qt/X/Chromium deps for Phase 3 (ref #127)
Phase 3 replaces the Phase 2 Chromium+PySide6 stack with a single-process
Pillow+/dev/fb0 kiosk. Dependency cleanup:

 dropped: chromium, openbox, xserver-xorg, xinit, unclutter, nginx-light,
          python3-pip, python3-pyside6 (Recommends), python3-qasync
 added:   python3-pil, python3-evdev

Helper FastAPI deps unchanged (carried forward from Phase 2).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:21:31 +02:00
gandalfandClaude Opus 4.7 220ce87c9e feat(remote-ui/square): secubox-square-kiosk.service unit + fb0 udev rule (ref #127)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:20:04 +02:00
gandalfandClaude Opus 4.7 c8fb8fb5c5 feat(remote-ui/square/kiosk): __main__.py event loop + smoke tests (ref #127)
Wires HelperClient + TransportManager + SimState + RingDashboard +
RightPanel + FrameBuffer into a 30 FPS event loop. Probes transport
every 30s, fetches metrics every 2s (falls back to SIM), composes
800x480 frame (dashboard + panel), blits to /dev/fb0.

Touch input integration deferred; modules from Task 6 are available
for the bench iteration to wire when the hardware ships.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:16:26 +02:00
gandalfandClaude Opus 4.7 a61234db78 feat(remote-ui/square/kiosk): ring_dashboard.py with easing + alerts ribbon (ref #127)
The implementer adapted update_metrics() to store raw metric values in
_target/_current (e.g. cpu_percent=80.0) rather than the 0..1 extracted
ratio from the plan's code block — this matches the test's assertion
(_target["cpu_percent"] == 80.0). Module.extract() is now applied at
draw() time. Behaviour is identical from the renderer's perspective.

Also added 'if m.metric in metrics' guard so partial-metric updates
preserve existing targets for unspecified keys (kiosk-friendly).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:12:57 +02:00
gandalfandClaude Opus 4.7 4ac50ddeb3 feat(remote-ui/square/kiosk): right_panel.py tab manager with TDD (ref #127)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:08:25 +02:00
gandalfandClaude Opus 4.7 9e9beafcb3 feat(remote-ui/square/kiosk): tabs/mode_controls.py with TDD (ref #127)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:05:20 +02:00
gandalfandClaude Opus 4.7 f26f6d9f24 feat(remote-ui/square/kiosk): tabs/console.py with TDD (ref #127)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 22:02:33 +02:00
gandalfandClaude Opus 4.7 4600f51cba feat(remote-ui/square/kiosk): tabs/module_detail.py with TDD (ref #127)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 21:59:41 +02:00
gandalf 2b9c6345f8 feat(remote-ui/square/kiosk): tabs/alerts.py with TDD (ref #127) 2026-05-13 20:22:20 +02:00
gandalf 9f7623b0d7 feat(remote-ui/square/kiosk): helper_client.py sync httpx UDS with TDD (ref #127) 2026-05-13 20:19:42 +02:00
gandalf 590cd8e069 feat(remote-ui/square/kiosk): transport_manager.py with TDD (ref #127) 2026-05-13 18:03:42 +02:00
gandalf 11d9987093 feat(remote-ui/square/kiosk): touch_input.py evdev reader + classify (ref #127) 2026-05-13 17:55:47 +02:00
gandalf 3ac33bfa3e feat(remote-ui/square/kiosk): framebuffer.py mmap helper with TDD (ref #127) 2026-05-13 17:50:47 +02:00
gandalf f5b5335ff8 feat(remote-ui/square/kiosk): sim.py drift generator with TDD (ref #127) 2026-05-13 17:45:40 +02:00
gandalf 48a003d2d1 feat(remote-ui/square/kiosk): theme.py + modules_table.py with TDD (ref #127) 2026-05-13 17:32:46 +02:00
gandalf 08b59a81e1 feat(remote-ui/square): scaffold kiosk/ package skeleton (ref #127) 2026-05-13 17:26:47 +02:00
gandalfandClaude Opus 4.7 53df9b6014 docs(remote-ui): Phase 3 implementation plan — Pillow+framebuffer kiosk (ref #127)
25 tasks covering kiosk skeleton, theme + modules table, sim, framebuffer +
touch input, transport manager + helper client, 4 right-pane tabs (Alerts /
Module Detail / Console / Mode Controls), right_panel composer, ring_dashboard
(Pillow port of Phase 2's QPainter ring renderer — visually faithful to
round/'s dashboard but independently authored, since round/fb_dashboard.py
must remain unchanged), __main__ event loop, systemd unit, fb udev rule,
debian/control simplification (drop Qt/X/Chromium, add python3-pil/evdev),
build script + firstboot + deploy.sh edits, README + CLAUDE.md docs,
regression test, Pi 4B + Pi 400 hardware bench tests, open PR.

Plan supersedes Phase 2 PR #131 (closed). Phase 2 helper FastAPI carries
forward unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 17:20:01 +02:00
gandalfandClaude Opus 4.7 12462a5230 feat(remote-ui/square): carry forward Phase 2 helper + debian + firstboot (ref #127)
Snapshot of Phase 2 PR #131's validated infrastructure, brought into Phase 3
unmodified. The Chromium+PySide6 dual-window kiosk path is dropped (replaced
by Pillow+framebuffer per the Phase 3 design spec). What carries forward:

packages/secubox-eye-square/
├── helper/                  21 pytest cases green
│   ├── eye_square_helper/   FastAPI app + SO_PEERCRED auth + 4 route modules
│   │   ├── app.py           middleware (JSONResponse fix preserved)
│   │   ├── auth.py          ALLOWED_UIDS frozenset
│   │   ├── __main__.py      uvicorn UDS bind
│   │   └── routes/          usb_gadget, service, lockdown, console
│   └── tests/               6 test files (test_e2e.py dropped — it depended
│                            on the right_panel package which is gone)
└── debian/                  arm64 package shell; control will be edited by
                             Phase 3 plan to drop Chromium/Qt/X deps

remote-ui/square/
├── files/etc/systemd/system/
│   ├── secubox-eye-square-helper.service    Helper FastAPI as secubox-eye-square user
│   ├── secubox-otg-gadget.service           configfs composite gadget (square variant)
│   └── secubox-firstboot.service            oneshot trigger for firstboot.sh
├── files/etc/udev/rules.d/
│   └── 90-secubox-otg-square.rules          host-side interface rename rule
├── files/etc/apparmor.d/
│   └── secubox-eye-square-helper            CAP_NET_ADMIN + CAP_SYS_ADMIN scope
├── files/etc/secubox/eye-square.toml.example
├── files/usr/local/sbin/firstboot.sh        GPIO 5V check + hostname + SSH + toml
├── build-eye-square-image.sh                will be heavily modified — drop Chromium/Qt/X
├── install_pi4.sh                           SD flash with safety guards
└── deploy.sh                                SSH hot-update — kiosk service list will change

Phase 1's remote-ui/common/ (PR #130) is NOT brought in — Phase 3 doesn't
consume it. round/fb_dashboard.py stays on master untouched.

What's DROPPED from Phase 2 (not brought in):
- right_panel/ entire PySide6 widget tree + ipc_bridge + WebSocket server
- square-bridge.js Chromium-side TM hook override
- secubox-kiosk-x.service, secubox-square-chromium.service,
  secubox-square-right-panel.service systemd units
- etc/openbox/{autostart,rc.xml}, etc/nginx/sites-available/secubox-square
- home/secubox/.xinitrc

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 16:58:17 +02:00
70 changed files with 6661 additions and 0 deletions
File diff suppressed because it is too large Load Diff
+4
View File
@@ -0,0 +1,4 @@
__pycache__/
*.py[cod]
*$py.class
.pytest_cache/
@@ -0,0 +1,8 @@
secubox-eye-square (1.0.0-1~bookworm1) bookworm; urgency=medium
* Initial release. Phase 2 of issue #127.
* Dual-pane kiosk: Chromium left, PySide6 right.
* Helper FastAPI on Unix socket with SO_PEERCRED.
* Variant-aware USB gadget composite.
-- Gerald KERMA <devel@cybermind.fr> Wed, 13 May 2026 00:00:00 +0200
@@ -0,0 +1 @@
13
@@ -0,0 +1,29 @@
Source: secubox-eye-square
Section: admin
Priority: optional
Maintainer: Gerald KERMA <devel@cybermind.fr>
Build-Depends: debhelper-compat (= 13), dh-python, python3-all
Standards-Version: 4.6.2
Package: secubox-eye-square
Architecture: arm64
Depends:
${misc:Depends},
${python3:Depends},
secubox-core,
secubox-eye-remote,
python3-pil,
python3-evdev,
python3-fastapi,
python3-uvicorn,
python3-websockets,
python3-httpx,
apparmor-utils
Description: SecuBox Eye Remote — Square variant (Pi 4B / Pi 400 + 7" 800x480)
Pillow-on-framebuffer single-process kiosk. Renders the SecuBox dashboard
directly to /dev/fb0 — no X server, no Qt, no Chromium. Companion to the
round/ Pi Zero W variant (also Pillow+fb).
.
Includes a privileged Helper FastAPI on a Unix socket (SO_PEERCRED) for
USB gadget mode switching, service restart, lockdown (nftables atomic
swap), and console streaming.
+28
View File
@@ -0,0 +1,28 @@
#!/bin/sh
set -e
case "$1" in
configure)
# Ensure system user exists
if ! id secubox-eye-square >/dev/null 2>&1; then
useradd --system --no-create-home --shell /usr/sbin/nologin secubox-eye-square
fi
# Ensure runtime + audit dirs
mkdir -p /run/secubox /var/log/secubox
chown secubox-eye-square:secubox-eye-square /run/secubox /var/log/secubox
# Activate AppArmor profile
if command -v apparmor_parser >/dev/null 2>&1; then
apparmor_parser -r /etc/apparmor.d/secubox-eye-square-helper || true
fi
# Reload systemd
systemctl daemon-reload || true
systemctl enable secubox-eye-square-helper.service || true
systemctl enable secubox-otg-gadget.service || true
;;
esac
#DEBHELPER#
exit 0
+14
View File
@@ -0,0 +1,14 @@
#!/bin/sh
set -e
case "$1" in
remove|upgrade|deconfigure)
systemctl stop secubox-square-chromium.service || true
systemctl stop secubox-square-right-panel.service || true
systemctl stop secubox-eye-square-helper.service || true
systemctl stop secubox-otg-gadget.service || true
;;
esac
#DEBHELPER#
exit 0
+14
View File
@@ -0,0 +1,14 @@
#!/usr/bin/make -f
%:
dh $@ --with python3 --buildsystem=pybuild
override_dh_auto_install:
dh_auto_install
# Helper package
mkdir -p debian/secubox-eye-square/usr/lib/python3/dist-packages/eye_square_helper
cp -r helper/eye_square_helper/. debian/secubox-eye-square/usr/lib/python3/dist-packages/eye_square_helper/
# Right-panel package
mkdir -p debian/secubox-eye-square/usr/lib/python3/dist-packages/secubox_eye_square_right_panel
cp -r right_panel/secubox_eye_square_right_panel/. debian/secubox-eye-square/usr/lib/python3/dist-packages/secubox_eye_square_right_panel/
# Config files (systemd units, nginx site, etc) from the sibling remote-ui/square/files/
cp -r ../../remote-ui/square/files/. debian/secubox-eye-square/
@@ -0,0 +1,33 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Entry point: bind FastAPI helper to /run/secubox/eye-square-helper.sock."""
from __future__ import annotations
import os
import sys
from pathlib import Path
import uvicorn
from eye_square_helper.app import app
SOCK = Path(os.environ.get("EYE_SQUARE_HELPER_SOCK", "/run/secubox/eye-square-helper.sock"))
def main() -> int:
SOCK.parent.mkdir(parents=True, exist_ok=True)
if SOCK.exists():
SOCK.unlink()
uvicorn.run(
app,
uds=str(SOCK),
log_level="info",
access_log=False,
)
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,88 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""FastAPI application factory for the eye-square helper.
Binds to a Unix socket (set in __main__.py) and rejects any client whose
SO_PEERCRED UID is not in ALLOWED_UIDS. Routers will be added in Tasks 3-5.
"""
from __future__ import annotations
import logging
import os
import socket
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from .auth import ALLOWED_UIDS, get_peer_uid
log = logging.getLogger("eye_square_helper")
def create_app() -> FastAPI:
"""Build the FastAPI app. Tasks 3-5 will add routers."""
app = FastAPI(
title="SecuBox Eye Square Helper",
description="Privileged local operations for remote-ui/square/. Unix-socket only, SO_PEERCRED-authenticated.",
version="0.1.0",
)
@app.middleware("http")
async def peercred_auth(request: Request, call_next):
# Best-effort peer-cred check. uvicorn over UDS exposes the underlying
# transport via request.scope['transport'] or request['transport'] in
# Starlette terms. We try a few paths to be resilient to version drift.
#
# NOTE: HTTPException raised inside @app.middleware("http") is NOT
# caught by FastAPI's exception handlers — it propagates as a 500.
# We must return a JSONResponse directly.
sock = _extract_peer_socket(request)
if sock is None:
log.warning("No transport socket on request; rejecting")
return JSONResponse(status_code=401, content={"detail": "cannot resolve peer"})
try:
uid = get_peer_uid(sock)
except OSError as e:
log.warning("SO_PEERCRED failed: %s", e)
return JSONResponse(status_code=401, content={"detail": "peer credentials unavailable"})
if uid not in ALLOWED_UIDS:
log.warning("Rejecting UID %d (allowed: %s)", uid, sorted(ALLOWED_UIDS))
return JSONResponse(status_code=403, content={"detail": "UID not allowed"})
return await call_next(request)
@app.get("/health")
async def health():
return {"status": "ok", "uid": os.getuid()}
from .routes.usb_gadget import router as usb_gadget_router
app.include_router(usb_gadget_router)
from .routes.service import router as service_router
from .routes.lockdown import router as lockdown_router
app.include_router(service_router)
app.include_router(lockdown_router)
from .routes.console import router as console_router
app.include_router(console_router)
return app
def _extract_peer_socket(request: Request) -> socket.socket | None:
"""Try several Starlette/uvicorn paths to find the connected peer socket."""
scope = request.scope
# uvicorn over UDS, modern Starlette: scope['extensions']['transport'] or similar
for key in ("transport", "asgi-transport"):
ts = scope.get(key)
if ts is not None and hasattr(ts, "get_extra_info"):
sock = ts.get_extra_info("socket")
if sock is not None:
return sock
# Newer uvicorn versions expose the socket via the underlying transport
# accessible through the protocol; the TestClient path won't have one.
return None
app = create_app()
@@ -0,0 +1,49 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""SO_PEERCRED-based authentication for the eye-square helper Unix socket.
The helper listens on /run/secubox/eye-square-helper.sock. Only specific UIDs
are allowed to call it (the dashboard user `secubox` and the right-panel user
`secubox-eye-square`, plus root for admin). Peer credentials are read from the
connected socket via SO_PEERCRED.
"""
from __future__ import annotations
import pwd
import socket
import struct
_ALLOWED_USERS = ("secubox", "secubox-eye-square", "root")
def _resolve_uid_by_name(name: str) -> int | None:
"""Resolve a username to its UID, or None if the user does not exist."""
try:
return pwd.getpwnam(name).pw_uid
except KeyError:
return None
def _build_allowed_uids() -> frozenset[int]:
"""Build the set of UIDs allowed to talk to the helper at module import."""
uids: set[int] = set()
for name in _ALLOWED_USERS:
uid = _resolve_uid_by_name(name)
if uid is not None:
uids.add(uid)
return frozenset(uids)
ALLOWED_UIDS: frozenset[int] = _build_allowed_uids()
def get_peer_uid(sock: socket.socket) -> int:
"""Return the UID of the peer on a connected AF_UNIX SOCK_STREAM socket.
Uses SO_PEERCRED to read the (pid, uid, gid) tuple. Linux-specific.
"""
creds = sock.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"))
_, uid, _ = struct.unpack("3i", creds)
return uid
@@ -0,0 +1,78 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Console WebSocket stream — tails /dev/ttyACM0 (satellite mode) or journalctl (kiosk mode)."""
from __future__ import annotations
import asyncio
import logging
import os
from pathlib import Path
from typing import AsyncIterator
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
router = APIRouter(prefix="/console", tags=["console"])
log = logging.getLogger("eye_square_helper.console")
TTY_DEVICE = os.environ.get("EYE_SQUARE_TTY_DEVICE", "/dev/ttyACM0")
TRANSPORT_STATE_FILE = Path(os.environ.get(
"EYE_SQUARE_TRANSPORT_STATE",
"/run/secubox/transport.state",
))
def _read_transport() -> str:
"""Read the current TransportManager state from the cache file."""
try:
return TRANSPORT_STATE_FILE.read_text().strip()
except OSError:
return "SIM"
def _select_source() -> str:
"""Choose source: tty device when OTG + present; otherwise journalctl."""
if _read_transport() == "OTG" and os.path.exists(TTY_DEVICE):
return TTY_DEVICE
return "journalctl"
async def _spawn_tail(source: str) -> AsyncIterator[str]:
"""Spawn tail process and yield decoded lines."""
if source == "journalctl":
cmd = ["journalctl", "-u", "secubox-*", "-f", "--no-pager", "-o", "short"]
else:
cmd = ["cat", source]
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
assert proc.stdout is not None
try:
while True:
line = await proc.stdout.readline()
if not line:
break
yield line.decode("utf-8", errors="replace").rstrip()
finally:
if proc.returncode is None:
try:
proc.terminate()
except ProcessLookupError:
pass
await proc.wait()
@router.websocket("/stream")
async def stream(ws: WebSocket):
"""Pump lines from the current source to the connected client."""
await ws.accept()
source = _select_source()
log.info("console stream source: %s", source)
try:
async for line in _spawn_tail(source):
await ws.send_text(line)
except WebSocketDisconnect:
log.debug("client disconnected")
@@ -0,0 +1,61 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Lockdown route — atomic nftables ruleset swap to deny-all. Audit-logged."""
from __future__ import annotations
import logging
import os
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
router = APIRouter(prefix="/lockdown", tags=["lockdown"])
log = logging.getLogger("eye_square_helper.lockdown")
LOCKDOWN_RULESET = os.environ.get(
"EYE_SQUARE_LOCKDOWN_RULESET",
"/etc/secubox/firewall/lockdown.nft",
)
AUDIT_LOG = Path(os.environ.get(
"EYE_SQUARE_AUDIT_LOG",
"/var/log/secubox/audit.log",
))
class LockdownRequest(BaseModel):
confirm: Literal["lockdown"]
def _apply_lockdown() -> tuple[str, int]:
"""Run `nft -f LOCKDOWN_RULESET` and audit-log the action."""
result = subprocess.run(
["nft", "-f", LOCKDOWN_RULESET],
capture_output=True,
text=True,
timeout=30,
)
stdout = result.stdout.strip() or result.stderr.strip()
rc = result.returncode
ts = datetime.now(timezone.utc).isoformat()
line = f"{ts} lockdown applied via eye-square-helper rc={rc}\n"
try:
AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True)
with AUDIT_LOG.open("a") as f:
f.write(line)
except OSError as e:
log.error("audit log write failed: %s", e)
return stdout, rc
@router.post("")
async def lockdown(request: LockdownRequest):
stdout, rc = _apply_lockdown()
if rc != 0:
raise HTTPException(status_code=500, detail=f"nft -f failed: {stdout}")
return {"applied": True, "exit_code": rc}
@@ -0,0 +1,52 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Service-restart route — allow-listed systemd unit names only."""
from __future__ import annotations
import subprocess
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
router = APIRouter(prefix="/service", tags=["service"])
# Strict allow-list. Match systemd unit names; no patterns, no wildcards.
ALLOWED_UNITS = frozenset({
"secubox-hub",
"secubox-auth",
"secubox-system",
"secubox-crowdsec",
"secubox-wireguard",
"secubox-dpi",
"secubox-dns",
})
# Unit-name regex: lowercase alnum + dot/dash/underscore, must start and end with alnum
_UNIT_PATTERN = r"^[a-z0-9][a-z0-9._-]*[a-z0-9]$"
class RestartRequest(BaseModel):
unit: str = Field(..., min_length=3, max_length=128, pattern=_UNIT_PATTERN)
def _run_systemctl(verb: str, unit: str) -> tuple[str, int]:
"""Run `systemctl <verb> <unit>`. Returns (stdout, rc)."""
result = subprocess.run(
["systemctl", verb, unit],
capture_output=True,
text=True,
timeout=30,
)
return result.stdout.strip(), result.returncode
@router.post("/restart")
async def restart_service(request: RestartRequest):
if request.unit not in ALLOWED_UNITS:
raise HTTPException(status_code=403, detail=f"unit '{request.unit}' not in allow-list")
stdout, rc = _run_systemctl("restart", request.unit)
if rc != 0:
raise HTTPException(status_code=500, detail=f"systemctl restart failed: {stdout}")
return {"unit": request.unit, "verb": "restart", "exit_code": rc}
@@ -0,0 +1,65 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""USB gadget mode switching routes.
Wraps invocations of secubox-otg-gadget.sh with VARIANT=square and
GADGET_NAME=secubox-square so the configfs gadget directory is variant-specific.
"""
from __future__ import annotations
import os
import subprocess
from typing import Literal
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
router = APIRouter(prefix="/usb-gadget", tags=["usb-gadget"])
GadgetMode = Literal["normal", "flash", "debug", "tty", "auth"]
_GADGET_SCRIPT = os.environ.get(
"EYE_SQUARE_GADGET_SCRIPT",
"/usr/local/sbin/secubox-otg-gadget.sh",
)
class ModeRequest(BaseModel):
mode: GadgetMode
def _run_gadget_script(argv: list[str]) -> tuple[str, int]:
"""Run secubox-otg-gadget.sh with GADGET_NAME=secubox-square. Returns (stdout, rc)."""
env = os.environ.copy()
env["GADGET_NAME"] = "secubox-square"
env["VARIANT"] = "square"
result = subprocess.run(
[_GADGET_SCRIPT, *argv],
env=env,
capture_output=True,
text=True,
timeout=30,
)
return result.stdout.strip(), result.returncode
@router.get("/state")
async def get_state():
"""Return the current USB gadget mode (parsed from script status)."""
stdout, rc = _run_gadget_script(["status"])
mode = "unknown"
for line in stdout.splitlines():
if line.lower().startswith("mode:"):
mode = line.split(":", 1)[1].strip()
break
return {"mode": mode, "exit_code": rc}
@router.post("/mode")
async def set_mode(request: ModeRequest):
"""Atomically switch the USB gadget composite to the requested mode."""
stdout, rc = _run_gadget_script([request.mode])
if rc != 0:
raise HTTPException(status_code=500, detail=f"gadget script failed: {stdout}")
return {"mode": request.mode, "exit_code": rc}
@@ -0,0 +1,8 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
import os
import sys
from pathlib import Path
_PKG = Path(__file__).resolve().parent.parent
if str(_PKG) not in sys.path:
sys.path.insert(0, str(_PKG))
@@ -0,0 +1,24 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Smoke tests for the helper FastAPI app."""
from __future__ import annotations
from fastapi.testclient import TestClient
from eye_square_helper.app import create_app
def test_app_constructs():
app = create_app()
assert app.title == "SecuBox Eye Square Helper"
def test_health_endpoint_responds():
"""TestClient runs over HTTP/TCP — there's no SO_PEERCRED on the wire,
so the middleware returns 401. That's correct behaviour for a non-UDS
transport: the helper MUST refuse non-Unix-socket peers."""
app = create_app()
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/health")
# 401 = peer-cred not available on TCP. The middleware did its job.
assert response.status_code == 401
@@ -0,0 +1,39 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for eye_square_helper.auth — SO_PEERCRED-based UID check."""
from __future__ import annotations
import os
import socket
import pytest
from eye_square_helper.auth import get_peer_uid, _resolve_uid_by_name
def test_get_peer_uid_returns_calling_uid(tmp_path):
"""Calling the helper from the same process must return os.getuid()."""
sock_path = tmp_path / "test.sock"
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server.bind(str(sock_path))
server.listen(1)
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client.connect(str(sock_path))
conn, _ = server.accept()
try:
uid = get_peer_uid(conn)
assert uid == os.getuid()
finally:
conn.close()
client.close()
server.close()
def test_resolve_uid_by_name_root_is_zero():
assert _resolve_uid_by_name("root") == 0
def test_resolve_uid_by_name_unknown_returns_none():
assert _resolve_uid_by_name("nonexistent-user-xyz-12345") is None
@@ -0,0 +1,55 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Tests for helper /console/stream WebSocket route."""
from __future__ import annotations
from unittest.mock import patch
from fastapi.testclient import TestClient
from eye_square_helper.app import create_app
from eye_square_helper.routes.console import _select_source
def test_select_source_satellite_returns_tty():
"""When TM state is OTG and /dev/ttyACM0 exists, return that path."""
with patch("eye_square_helper.routes.console._read_transport", return_value="OTG"), \
patch("os.path.exists", return_value=True):
assert _select_source() == "/dev/ttyACM0"
def test_select_source_kiosk_returns_journalctl():
"""When TM state is not OTG, return journalctl."""
with patch("eye_square_helper.routes.console._read_transport", return_value="WiFi"):
assert _select_source() == "journalctl"
def test_select_source_sim_returns_journalctl():
with patch("eye_square_helper.routes.console._read_transport", return_value="SIM"):
assert _select_source() == "journalctl"
def test_select_source_otg_but_no_tty_returns_journalctl():
"""If OTG but the tty device is missing, fall back to journalctl."""
with patch("eye_square_helper.routes.console._read_transport", return_value="OTG"), \
patch("os.path.exists", return_value=False):
assert _select_source() == "journalctl"
def test_console_ws_streams_lines():
"""WS connects, mocked tail yields 2 lines, client receives them."""
async def fake_tail(source):
yield "line A"
yield "line B"
app = create_app()
client = TestClient(app)
with patch("eye_square_helper.routes.console._spawn_tail", fake_tail), \
patch("eye_square_helper.routes.console._select_source", return_value="journalctl"):
with client.websocket_connect("/console/stream") as ws:
received = []
for _ in range(2):
received.append(ws.receive_text())
assert received == ["line A", "line B"]
@@ -0,0 +1,52 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Tests for helper /lockdown route — atomic nft ruleset swap."""
from __future__ import annotations
from unittest.mock import patch
import pytest
from fastapi.testclient import TestClient
from eye_square_helper.app import create_app
@pytest.fixture
def client():
return TestClient(create_app(), raise_server_exceptions=False)
def _auth_patches():
class _FakeSocket:
pass
sock_patch = patch("eye_square_helper.app._extract_peer_socket", return_value=_FakeSocket())
uid_patch = patch("eye_square_helper.app.get_peer_uid", return_value=0)
return sock_patch, uid_patch
def test_lockdown_applies_when_confirmed(client):
sock_p, uid_p = _auth_patches()
with sock_p, uid_p, patch("eye_square_helper.routes.lockdown._apply_lockdown") as mock_apply:
mock_apply.return_value = ("ruleset applied", 0)
response = client.post("/lockdown", json={"confirm": "lockdown"})
assert response.status_code == 200, response.text
body = response.json()
assert body["applied"] is True
assert body["exit_code"] == 0
def test_lockdown_requires_exact_confirm_string(client):
sock_p, uid_p = _auth_patches()
with sock_p, uid_p:
response = client.post("/lockdown", json={"confirm": "yes"})
assert response.status_code == 422 # Pydantic Literal rejection
def test_lockdown_propagates_nft_failure(client):
sock_p, uid_p = _auth_patches()
with sock_p, uid_p, patch("eye_square_helper.routes.lockdown._apply_lockdown") as mock_apply:
mock_apply.return_value = ("nft: syntax error", 1)
response = client.post("/lockdown", json={"confirm": "lockdown"})
assert response.status_code == 500
@@ -0,0 +1,63 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Tests for helper /service/restart route — allow-listed systemd units only."""
from __future__ import annotations
from unittest.mock import patch
import pytest
from fastapi.testclient import TestClient
from eye_square_helper.app import create_app
@pytest.fixture
def client():
return TestClient(create_app(), raise_server_exceptions=False)
def _auth_patches():
"""Mock both peer-cred extraction and UID lookup so requests pass middleware."""
class _FakeSocket:
pass
sock_patch = patch("eye_square_helper.app._extract_peer_socket", return_value=_FakeSocket())
uid_patch = patch("eye_square_helper.app.get_peer_uid", return_value=0)
return sock_patch, uid_patch
def test_restart_allowed_unit(client):
sock_p, uid_p = _auth_patches()
with sock_p, uid_p, patch("eye_square_helper.routes.service._run_systemctl") as mock_sysctl:
mock_sysctl.return_value = ("", 0)
response = client.post("/service/restart", json={"unit": "secubox-hub"})
assert response.status_code == 200, response.text
body = response.json()
assert body["unit"] == "secubox-hub"
assert body["verb"] == "restart"
assert body["exit_code"] == 0
def test_restart_rejects_arbitrary_unit(client):
sock_p, uid_p = _auth_patches()
with sock_p, uid_p:
# nginx is a real unit name but not in our allow-list
response = client.post("/service/restart", json={"unit": "nginx"})
assert response.status_code == 403
def test_restart_validation_rejects_injection_attempt(client):
sock_p, uid_p = _auth_patches()
with sock_p, uid_p:
# Semicolons / shell metacharacters MUST fail Pydantic pattern
response = client.post("/service/restart", json={"unit": "secubox-hub; rm -rf /"})
assert response.status_code == 422
def test_restart_propagates_systemctl_failure(client):
sock_p, uid_p = _auth_patches()
with sock_p, uid_p, patch("eye_square_helper.routes.service._run_systemctl") as mock_sysctl:
mock_sysctl.return_value = ("unit not found", 1)
response = client.post("/service/restart", json={"unit": "secubox-hub"})
assert response.status_code == 500
@@ -0,0 +1,75 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Tests for helper /usb-gadget routes."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from eye_square_helper.app import create_app
@pytest.fixture
def client():
app = create_app()
return TestClient(app, raise_server_exceptions=False)
def _auth_patches():
"""Return the pair of patches that let a test request pass the middleware.
_extract_peer_socket normally returns None for TestClient (TCP), which
causes an immediate 401. We patch it to return a dummy socket object so
the middleware proceeds to call get_peer_uid. Then we patch get_peer_uid
to return 0 (root), which is always in ALLOWED_UIDS.
"""
mock_sock = MagicMock()
patch_socket = patch("eye_square_helper.app._extract_peer_socket", return_value=mock_sock)
patch_uid = patch("eye_square_helper.app.get_peer_uid", return_value=0)
return patch_socket, patch_uid
def test_usb_gadget_state_returns_current_mode(client):
"""GET /usb-gadget/state returns the current gadget mode."""
patch_socket, patch_uid = _auth_patches()
with patch_socket, patch_uid, \
patch("eye_square_helper.routes.usb_gadget._run_gadget_script") as mock_run:
mock_run.return_value = ("Mode: normal\nUDC: 20980000.usb", 0)
response = client.get("/usb-gadget/state")
assert response.status_code == 200
body = response.json()
assert body["mode"] == "normal"
assert body["exit_code"] == 0
def test_usb_gadget_mode_accepts_valid_modes(client):
valid_modes = ["normal", "flash", "debug", "tty", "auth"]
for mode in valid_modes:
patch_socket, patch_uid = _auth_patches()
with patch_socket, patch_uid, \
patch("eye_square_helper.routes.usb_gadget._run_gadget_script") as mock_run:
mock_run.return_value = ("", 0)
response = client.post("/usb-gadget/mode", json={"mode": mode})
assert response.status_code == 200, f"mode {mode} rejected: {response.text}"
def test_usb_gadget_mode_rejects_unknown(client):
patch_socket, patch_uid = _auth_patches()
with patch_socket, patch_uid:
response = client.post("/usb-gadget/mode", json={"mode": "blorp"})
# Pydantic Literal validation → 422 Unprocessable Entity
assert response.status_code == 422
def test_usb_gadget_mode_propagates_script_failure(client):
"""If the gadget script returns nonzero, the endpoint should 500."""
patch_socket, patch_uid = _auth_patches()
with patch_socket, patch_uid, \
patch("eye_square_helper.routes.usb_gadget._run_gadget_script") as mock_run:
mock_run.return_value = ("ERROR: configfs not mounted", 1)
response = client.post("/usb-gadget/mode", json={"mode": "normal"})
assert response.status_code == 500
@@ -0,0 +1,106 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/__main__.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""SecuBox Eye Square kiosk — event loop driver."""
from __future__ import annotations
import logging
import os
import sys
import time
from pathlib import Path
from PIL import Image
from .framebuffer import FrameBuffer
from .helper_client import HelperClient
from .right_panel import RightPanel
from .ring_dashboard import RingDashboard
from .sim import SimState, step
from .transport_manager import TransportManager
log = logging.getLogger("secubox_eye_square_kiosk")
FB_PATH = os.environ.get("EYE_SQUARE_FB", "/dev/fb0")
HELPER_SOCK = os.environ.get("EYE_SQUARE_HELPER_SOCK",
"/run/secubox/eye-square-helper.sock")
TARGET_FPS = 30
PROBE_INTERVAL_S = 30
METRICS_INTERVAL_S = 2
def main() -> int:
logging.basicConfig(level=logging.INFO,
format="%(asctime)s %(name)s %(levelname)s %(message)s")
log.info("Starting SecuBox Eye Square kiosk")
# Helper + TransportManager
helper = HelperClient(HELPER_SOCK)
tm = TransportManager(simulate=False)
tm.probe()
# SIM state for fallback
sim = SimState()
# Dashboard + right panel
rd = RingDashboard()
panel = RightPanel(helper)
rd.on_module_tap = panel.on_module_tap
tm.on_transport_change = lambda active: (
panel.on_transport_change(active),
rd.set_transport(active),
)
# Framebuffer
try:
fb = FrameBuffer(FB_PATH)
except OSError as e:
log.error("Cannot open framebuffer %s: %s", FB_PATH, e)
return 1
last_probe = 0.0
last_metrics = 0.0
frame_period = 1.0 / TARGET_FPS
try:
while True:
now = time.time()
# Periodic transport probe
if now - last_probe > PROBE_INTERVAL_S:
tm.probe()
last_probe = now
# Periodic metrics fetch (or SIM drift)
if now - last_metrics > METRICS_INTERVAL_S:
metrics = tm.fetch_metrics()
if metrics is None:
step(sim, refresh_interval_s=METRICS_INTERVAL_S)
metrics = sim.to_dict()
rd.update_metrics(metrics)
last_metrics = now
# Animation tick
rd.advance()
# Compose frame
full = Image.new("RGBA", (800, 480), (0, 0, 0, 255))
full.paste(rd.draw(), (0, 0))
panel_img = Image.new("RGBA", (320, 480), (0, 0, 0, 255))
panel.draw(panel_img)
full.paste(panel_img, (480, 0))
fb.blit(full)
time.sleep(frame_period)
except KeyboardInterrupt:
log.info("Shutting down kiosk")
finally:
fb.close()
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,55 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/framebuffer.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Direct /dev/fb0 framebuffer blit via mmap.
The Pi 4B's DSI panel exposes /dev/fb0 at 800×480, 32-bit BGRA when the
vc4-kms-v3d overlay is active. We open it once, mmap the full size, and
blit Pillow images into it on each render tick.
"""
from __future__ import annotations
import logging
import mmap
import os
from pathlib import Path
from PIL import Image
log = logging.getLogger("secubox_eye_square_kiosk.framebuffer")
class FrameBuffer:
"""Owns the mmap handle to /dev/fb0. Single-instance-per-process."""
def __init__(self, path: str = "/dev/fb0", width: int = 800, height: int = 480, bpp: int = 4):
self.path = path
self.width = width
self.height = height
self.bpp = bpp
self.size = width * height * bpp
self.fd = os.open(path, os.O_RDWR)
self.fb = mmap.mmap(self.fd, self.size, mmap.MAP_SHARED, mmap.PROT_WRITE)
def blit(self, image: Image.Image) -> None:
"""Push a Pillow image to the framebuffer. Image must be RGBA at exact resolution."""
if image.size != (self.width, self.height):
raise ValueError(
f"image size {image.size} doesn't match framebuffer {self.width}x{self.height}"
)
# Convert Pillow RGBA → BGRA for vc4-kms-v3d's little-endian BGRA32 layout
bgra = image.tobytes("raw", "BGRA")
self.fb.seek(0)
self.fb.write(bgra)
def close(self) -> None:
self.fb.close()
os.close(self.fd)
def __enter__(self) -> "FrameBuffer":
return self
def __exit__(self, *_args) -> None:
self.close()
@@ -0,0 +1,55 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/helper_client.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Sync httpx client for the eye-square-helper FastAPI Unix socket.
The kiosk is a single-threaded Python process; the helper exposes
privileged ops over /run/secubox/eye-square-helper.sock. All calls are
synchronous and propagate httpx exceptions to the caller.
"""
from __future__ import annotations
import logging
from typing import Any
import httpx
log = logging.getLogger("secubox_eye_square_kiosk.helper_client")
class HelperClient:
"""Calls the privileged helper. Construct once per process; thread-safe enough for kiosk."""
def __init__(self, socket_path: str, timeout: float = 10.0):
self.socket_path = socket_path
self.timeout = timeout
def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]:
transport = httpx.HTTPTransport(uds=self.socket_path)
with httpx.Client(transport=transport, base_url="http://localhost",
timeout=self.timeout) as c:
r = c.post(path, json=payload)
r.raise_for_status()
return r.json()
def _get(self, path: str) -> dict[str, Any]:
transport = httpx.HTTPTransport(uds=self.socket_path)
with httpx.Client(transport=transport, base_url="http://localhost",
timeout=self.timeout) as c:
r = c.get(path)
r.raise_for_status()
return r.json()
def set_usb_mode(self, mode: str) -> dict[str, Any]:
return self._post("/usb-gadget/mode", {"mode": mode})
def get_usb_state(self) -> dict[str, Any]:
return self._get("/usb-gadget/state")
def restart_service(self, unit: str) -> dict[str, Any]:
return self._post("/service/restart", {"unit": unit})
def lockdown(self) -> dict[str, Any]:
return self._post("/lockdown", {"confirm": "lockdown"})
@@ -0,0 +1,82 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""6-module RINGS table — colour, ring radius, metric extractor.
Hamiltonian order: AUTH → WALL → BOOT → MIND → ROOT → MESH → AUTH.
Each entry corresponds to one concentric arc on the 480×480 round canvas.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable
from . import theme
@dataclass(frozen=True)
class Module:
"""One module's rendering metadata."""
name: str # "AUTH", "WALL", ...
colour: tuple[int, int, int] # RGB tuple from theme.py
radius: int # arc radius in pixels (centre at 240,240)
metric: str # API field name, e.g. "cpu_percent"
unit: str # display unit, e.g. "%"
extract: Callable[[dict], float] # (state-dict) → 0..1 fill ratio
def _clamp(v: float, lo: float = 0.0, hi: float = 1.0) -> float:
return max(lo, min(hi, v))
MODULES: list[Module] = [
Module(
name="AUTH",
colour=theme.AUTH,
radius=214,
metric="cpu_percent",
unit="%",
extract=lambda s: _clamp(s.get("cpu_percent", 0.0) / 100.0),
),
Module(
name="WALL",
colour=theme.WALL,
radius=201,
metric="mem_percent",
unit="%",
extract=lambda s: _clamp(s.get("mem_percent", 0.0) / 100.0),
),
Module(
name="BOOT",
colour=theme.BOOT,
radius=188,
metric="disk_percent",
unit="%",
extract=lambda s: _clamp(s.get("disk_percent", 0.0) / 100.0),
),
Module(
name="MIND",
colour=theme.MIND,
radius=175,
metric="load_avg_1",
unit="×",
extract=lambda s: _clamp(s.get("load_avg_1", 0.0) / 4.0),
),
Module(
name="ROOT",
colour=theme.ROOT,
radius=162,
metric="cpu_temp",
unit="°C",
extract=lambda s: _clamp((s.get("cpu_temp", 35.0) - 35.0) / 50.0),
),
Module(
name="MESH",
colour=theme.MESH,
radius=149,
metric="wifi_rssi",
unit="dBm",
extract=lambda s: _clamp((s.get("wifi_rssi", -90) + 90.0) / 70.0),
),
]
@@ -0,0 +1,90 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/right_panel.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Right panel — tab bar at top, content area below. Owns 4 tab widgets."""
from __future__ import annotations
from PIL import Image, ImageDraw
from . import theme
from .tabs.alerts import AlertsTab
from .tabs.console import ConsoleTab
from .tabs.mode_controls import ModeControlsTab
from .tabs.module_detail import ModuleDetailTab
TAB_BAR_HEIGHT = 56
TAB_WIDTH = 80
TAB_LABELS = [
("alerts", "ALERTS"),
("module_detail", "DETAIL"),
("console", "CON"),
("mode_controls", "CTL"),
]
class RightPanel:
"""Manages the 4 tabs and the tab bar."""
def __init__(self, helper_client):
self.tabs = {
"alerts": AlertsTab(),
"module_detail": ModuleDetailTab(),
"console": ConsoleTab(),
"mode_controls": ModeControlsTab(helper_client),
}
# Wire alert tap → switch to module detail
self.tabs["alerts"].on_row_tap = self._on_alert_tapped
self.active_tab = "alerts"
def set_active_tab(self, name: str) -> None:
if name in self.tabs:
self.active_tab = name
def on_module_tap(self, module_name: str) -> None:
"""Called by ring_dashboard when the user taps a pod."""
self.active_tab = "module_detail"
# value/history can be empty for the initial switch; ring_dashboard will refresh
self.tabs["module_detail"].load_module(module_name, "", value=0.0, history=[])
def on_transport_change(self, active: str) -> None:
self.tabs["mode_controls"].update_transport(active)
def append_console_line(self, line: str) -> None:
self.tabs["console"].append_line(line)
def set_alerts(self, items) -> None:
self.tabs["alerts"].set_alerts(items)
def _on_alert_tapped(self, item) -> None:
self.active_tab = "module_detail"
self.tabs["module_detail"].load_module(item.module, "", value=0.0, history=[])
def handle_tap(self, x: int, y: int) -> None:
# Tab bar?
if y < TAB_BAR_HEIGHT:
tab_idx = x // TAB_WIDTH
if 0 <= tab_idx < len(TAB_LABELS):
self.active_tab = TAB_LABELS[tab_idx][0]
return
# Route to active tab (subtract tab bar offset)
self.tabs[self.active_tab].handle_tap(x, y - TAB_BAR_HEIGHT)
def draw(self, region: Image.Image) -> None:
"""Render tab bar + active tab into the 320x480 region."""
draw = ImageDraw.Draw(region)
w, h = region.size
# Tab bar background
draw.rectangle((0, 0, w, TAB_BAR_HEIGHT), fill=theme.COSMOS_BLACK)
for i, (key, label) in enumerate(TAB_LABELS):
x = i * TAB_WIDTH
colour = theme.GOLD_HERMETIC if key == self.active_tab else theme.TEXT_MUTED
draw.rectangle((x, 0, x + TAB_WIDTH, TAB_BAR_HEIGHT - 1),
outline=colour, width=1 if key != self.active_tab else 2)
draw.text((x + 8, 20), label, fill=colour)
# Content area
content_h = h - TAB_BAR_HEIGHT
content = Image.new("RGBA", (w, content_h), (0, 0, 0, 255))
self.tabs[self.active_tab].draw(content)
region.paste(content, (0, TAB_BAR_HEIGHT))
@@ -0,0 +1,148 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/ring_dashboard.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Ring dashboard — left 480x480 Pillow renderer.
Pixel-faithful intent vs Phase 1 round/index.html: 6 concentric arcs
(radii 214/201/188/175/162/149), each module colour-mapped, smooth fill
animation toward target value, central clock + hostname + uptime,
transport badge, status row, temperature bar. Alerts ribbon overlays
bottom 24px when severity warn.
"""
from __future__ import annotations
import math
import socket
import time
from datetime import datetime
from typing import Callable, Optional
from PIL import Image, ImageDraw
from . import theme
from .modules_table import MODULES, Module
CX, CY = 240, 240
RING_WIDTH = 5
EASE_STEPS = 8 # animation frames between metric updates
POD_DISTANCE = 235
ALERT_RIBBON_HEIGHT = 24
TRANSPORT_BADGE_Y = 14
class RingDashboard:
"""480x480 left half. update_metrics() sets target values; advance() eases
current values toward target each tick; draw() renders the frame."""
def __init__(self):
self.size = (480, 480)
self.transport = "SIM"
self.hostname = socket.gethostname()
self._current: dict[str, float] = {m.metric: 0.0 for m in MODULES}
self._target: dict[str, float] = {m.metric: 0.0 for m in MODULES}
self._alert_text = ""
self._alert_severity = "info"
self.on_module_tap: Callable[[str], None] = lambda _: None
def update_metrics(self, metrics: dict) -> None:
"""Set new target values. _current eases toward _target over EASE_STEPS frames.
Stores raw metric values (e.g. cpu_percent=80.0). The modules_table
extract() function converts to 0..1 fill ratio at draw time.
"""
for m in MODULES:
if m.metric in metrics:
self._target[m.metric] = float(metrics[m.metric])
def advance(self) -> None:
"""One easing frame — move _current toward _target by 1/EASE_STEPS."""
for m in MODULES:
cur = self._current[m.metric]
tgt = self._target[m.metric]
self._current[m.metric] = cur + (tgt - cur) / EASE_STEPS
def set_transport(self, active: str) -> None:
self.transport = active
def set_alert_ribbon(self, text: str, severity: str = "info") -> None:
self._alert_text = text
self._alert_severity = severity
def clear_alert_ribbon(self) -> None:
self._alert_text = ""
def handle_tap(self, x: int, y: int) -> None:
"""Detect pod taps. Pods sit at angles -π/2, -π/2+π/3, ... around the ring."""
dx, dy = x - CX, y - CY
dist = math.hypot(dx, dy)
if abs(dist - POD_DISTANCE) > 30:
return
# angle in radians, 0 = right, -π/2 = top
angle = math.atan2(dy, dx)
# Normalise so AUTH is at -π/2 (top), increment by π/3 clockwise
normalised = (angle + math.pi / 2) % (2 * math.pi)
idx = int(normalised / (math.pi / 3))
if 0 <= idx < len(MODULES):
self.on_module_tap(MODULES[idx].name)
def _pod_position(self, idx: int) -> tuple[int, int]:
"""Where to draw the idx-th pod's icon/label."""
angle = -math.pi / 2 + idx * (math.pi / 3)
x = CX + int(POD_DISTANCE * math.cos(angle))
y = CY + int(POD_DISTANCE * math.sin(angle))
return x, y
def draw(self) -> Image.Image:
img = Image.new("RGBA", self.size, theme.COSMOS_BLACK + (255,))
draw = ImageDraw.Draw(img)
# 6 rings — draw track then fill arc for each module
for m in MODULES:
pct = m.extract(self._current)
# ring track (full circle, very dark)
draw.arc(
(CX - m.radius, CY - m.radius, CX + m.radius, CY + m.radius),
start=-90, end=270,
fill=(0x14, 0x14, 0x14, 255), width=RING_WIDTH + 2,
)
# ring fill (proportional arc from top, clockwise)
if pct > 0.005:
end_angle = -90 + 360 * pct
draw.arc(
(CX - m.radius, CY - m.radius, CX + m.radius, CY + m.radius),
start=-90, end=end_angle,
fill=m.colour + (255,), width=RING_WIDTH,
)
# Pods — coloured dot at ring perimeter + module name label
for i, m in enumerate(MODULES):
px, py = self._pod_position(i)
# Coloured dot
draw.ellipse((px - 5, py - 5, px + 5, py + 5), fill=m.colour + (255,))
# Module name label below dot
draw.text((px - 16, py + 8), m.name, fill=theme.TEXT_PRIMARY)
# Central clock + hostname
now = datetime.now().strftime("%H:%M:%S")
date = datetime.now().strftime("%a %d %b")
draw.text((CX - 50, CY - 18), now, fill=theme.TEXT_PRIMARY)
draw.text((CX - 30, CY + 4), date, fill=theme.TEXT_MUTED)
draw.text((CX - 70, CY + 22), self.hostname[:18], fill=theme.TEXT_MUTED)
# Transport badge top-right
dot = "" if self.transport in ("OTG", "WiFi") else ""
dot_colour = theme.MATRIX_GREEN if dot == "" else theme.TEXT_MUTED
draw.text((CX + 110, TRANSPORT_BADGE_Y), f"{dot} {self.transport}",
fill=dot_colour)
# Alerts ribbon — overlay bottom 24px when alert is active
if self._alert_text:
ribbon_colour = theme.SEVERITY.get(self._alert_severity, theme.TEXT_MUTED)
draw.rectangle((0, 480 - ALERT_RIBBON_HEIGHT, 480, 480),
fill=theme.COSMOS_BLACK + (200,))
draw.text((10, 480 - ALERT_RIBBON_HEIGHT + 4),
f"{self._alert_text}"[:50], fill=ribbon_colour)
return img
@@ -0,0 +1,47 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/sim.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Simulation drift generator — Python port of Phase 1 sim.js.
When no SecuBox host responds, the kiosk uses these synthetic values so
the rings still animate plausibly. Random walk bounded to realistic ranges.
"""
from __future__ import annotations
import random
from dataclasses import dataclass, asdict
@dataclass
class SimState:
"""Drift state — mutable, advanced by step()."""
cpu_percent: float = 14.0
mem_percent: float = 42.0
disk_percent: float = 28.0
wifi_rssi: int = -63
load_avg_1: float = 0.18
cpu_temp: float = 44.0
uptime_seconds: float = 0.0
hostname: str = "secubox-zero"
def to_dict(self) -> dict:
return asdict(self)
def _walk(value: float, drift: float, lo: float, hi: float) -> float:
"""One step of a bounded random walk."""
new_value = value + (random.random() - 0.5) * drift
return max(lo, min(hi, new_value))
def step(state: SimState, refresh_interval_s: float = 2.0) -> None:
"""Advance state in place. refresh_interval_s adds to uptime."""
state.cpu_percent = _walk(state.cpu_percent, 12.0, 0.0, 100.0)
state.mem_percent = _walk(state.mem_percent, 3.0, 20.0, 95.0)
state.disk_percent = _walk(state.disk_percent, 0.7, 5.0, 95.0)
state.wifi_rssi = int(_walk(float(state.wifi_rssi), 5.0, -90.0, -20.0))
state.load_avg_1 = _walk(state.load_avg_1, 0.12, 0.0, 4.0)
state.cpu_temp = _walk(state.cpu_temp, 1.5, 35.0, 82.0)
state.uptime_seconds += refresh_interval_s
@@ -0,0 +1,79 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/tabs/alerts.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Alerts tab — Pillow-drawn scrollable list of recent system alerts."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, Optional
from PIL import Image, ImageDraw
from .. import theme
ROW_HEIGHT = 32
DOT_RADIUS = 4
TEXT_PAD_LEFT = 24
TEXT_PAD_RIGHT = 8
@dataclass
class AlertItem:
severity: str # "info" | "warn" | "crit"
time: str
module: str
message: str
class AlertsTab:
"""Scrollable list. Tap a row to drill into module detail."""
def __init__(self):
self.items: list[AlertItem] = []
self.scroll_offset = 0
self.on_row_tap: Callable[[AlertItem], None] = lambda _: None
def set_alerts(self, items: list[AlertItem]) -> None:
self.items = list(items)
self.scroll_offset = 0
def handle_tap(self, x: int, y: int) -> None:
"""Convert a tap at (x, y) (region-local coords) to a row hit."""
row_index = (y + self.scroll_offset) // ROW_HEIGHT
if 0 <= row_index < len(self.items):
self.on_row_tap(self.items[row_index])
def handle_drag(self, dx: int, dy: int) -> None:
"""Drag down scrolls list up (negative dy = scroll up)."""
self.scroll_offset = max(
0,
min(
max(0, len(self.items) * ROW_HEIGHT - 424),
self.scroll_offset - dy,
),
)
def draw(self, region: Image.Image) -> None:
"""Render alerts into the region (320x424 RGBA image)."""
draw = ImageDraw.Draw(region)
if not self.items:
draw.text((10, 10), "● NOMINAL", fill=theme.MATRIX_GREEN)
return
w, h = region.size
for i, item in enumerate(self.items):
y = i * ROW_HEIGHT - self.scroll_offset
if y + ROW_HEIGHT < 0 or y > h:
continue
dot = theme.SEVERITY.get(item.severity, theme.TEXT_MUTED)
draw.ellipse(
(8, y + 12, 8 + 2 * DOT_RADIUS, y + 12 + 2 * DOT_RADIUS),
fill=dot,
)
txt = f"{item.time} {item.module} {item.message}"
draw.text((TEXT_PAD_LEFT, y + 8), txt[:38],
fill=theme.TEXT_PRIMARY)
# divider line
draw.line((0, y + ROW_HEIGHT - 1, w, y + ROW_HEIGHT - 1),
fill=theme.TEXT_MUTED)
@@ -0,0 +1,55 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/tabs/console.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Console tab — text scrollback with a Freeze toggle."""
from __future__ import annotations
from PIL import Image, ImageDraw
from .. import theme
LINE_HEIGHT = 14
TOP_MARGIN = 8
BUTTON_Y = 380
BUTTON_HEIGHT = 32
BUTTON_X = 240
class ConsoleTab:
"""Read-only console tail. append_line() is no-op when frozen."""
def __init__(self, max_lines: int = 200):
self.lines: list[str] = []
self.frozen = False
self.max_lines = max_lines
def append_line(self, line: str) -> None:
if self.frozen:
return
self.lines.append(line)
if len(self.lines) > self.max_lines:
self.lines = self.lines[-self.max_lines:]
def handle_tap(self, x: int, y: int) -> None:
"""Tap on the Freeze button (bottom-right)?"""
if BUTTON_X <= x <= 320 and BUTTON_Y <= y <= BUTTON_Y + BUTTON_HEIGHT:
self.frozen = not self.frozen
def draw(self, region: Image.Image) -> None:
draw = ImageDraw.Draw(region)
w, h = region.size
# Background — solid black for readability
draw.rectangle((0, 0, w, h), fill=(0, 0, 0, 255))
# Render the last N lines that fit
visible_rows = (h - 50) // LINE_HEIGHT
for i, line in enumerate(self.lines[-visible_rows:]):
y = TOP_MARGIN + i * LINE_HEIGHT
draw.text((4, y), line[:48], fill=theme.MATRIX_GREEN)
# Freeze button
btn_label = "Resume" if self.frozen else "Freeze"
btn_fill = theme.GOLD_HERMETIC if self.frozen else theme.TEXT_MUTED
draw.rectangle((BUTTON_X, BUTTON_Y, w - 4, BUTTON_Y + BUTTON_HEIGHT),
outline=btn_fill, width=1)
draw.text((BUTTON_X + 8, BUTTON_Y + 8), btn_label, fill=btn_fill)
@@ -0,0 +1,138 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/tabs/mode_controls.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Mode Controls tab — USB gadget mode + service restart + lockdown + transport."""
from __future__ import annotations
import logging
from typing import Optional
from PIL import Image, ImageDraw
from .. import theme
log = logging.getLogger("secubox_eye_square_kiosk.tabs.mode_controls")
USB_BUTTONS = ["normal", "flash", "debug", "tty", "auth", "stop"]
SERVICE_BUTTONS = [
("secubox-hub", "RESTART HUB"),
("secubox-auth", "RESTART AUTH"),
("restart-all", "RESTART ALL"),
("lockdown", "LOCKDOWN !"),
]
DESTRUCTIVE = {"flash", "stop", "restart-all", "lockdown"}
USB_ROW_Y = 40
SERVICE_ROW_Y = 200
TRANSPORT_ROW_Y = 360
CELL_W = 100
CELL_H = 64
class ModeControlsTab:
"""Touch-button grid. Destructive actions require confirm tap."""
def __init__(self, helper_client):
self.helper = helper_client
self.transport_active = "SIM"
self.pending_confirm: Optional[str] = None
def update_transport(self, active: str) -> None:
self.transport_active = active
def handle_tap(self, x: int, y: int) -> None:
"""Map (x, y) to a button. Destructive actions stage pending_confirm; second tap commits."""
# USB mode buttons — top 2x3 grid
if USB_ROW_Y <= y < USB_ROW_Y + 2 * CELL_H:
row = (y - USB_ROW_Y) // CELL_H
col = x // CELL_W
idx = row * 3 + col
if 0 <= idx < len(USB_BUTTONS):
mode = USB_BUTTONS[idx]
self._invoke_or_stage(mode, lambda: self.helper.set_usb_mode(mode))
return
# Service buttons — middle 2x2 grid
if SERVICE_ROW_Y <= y < SERVICE_ROW_Y + 2 * CELL_H:
row = (y - SERVICE_ROW_Y) // CELL_H
col = x // (320 // 2)
idx = row * 2 + col
if 0 <= idx < len(SERVICE_BUTTONS):
action, _label = SERVICE_BUTTONS[idx]
self._invoke_or_stage(action, lambda: self._service_action(action))
return
def _invoke_or_stage(self, action: str, callback) -> None:
if action in DESTRUCTIVE and self.pending_confirm != action:
self.pending_confirm = action
return
self.pending_confirm = None
try:
callback()
except Exception as e:
log.warning("action %s failed: %s", action, e)
def confirm_pending(self) -> None:
"""Called externally after the user confirms via the confirm overlay tap."""
if self.pending_confirm is None:
return
action = self.pending_confirm
self.pending_confirm = None
try:
if action in USB_BUTTONS:
self.helper.set_usb_mode(action)
else:
self._service_action(action)
except Exception as e:
log.warning("confirm action %s failed: %s", action, e)
def _service_action(self, action: str) -> None:
if action == "lockdown":
self.helper.lockdown()
elif action == "restart-all":
for unit in ("secubox-hub", "secubox-auth", "secubox-system"):
self.helper.restart_service(unit)
else:
self.helper.restart_service(action)
def draw(self, region: Image.Image) -> None:
draw = ImageDraw.Draw(region)
w, _ = region.size
# USB buttons header
draw.text((10, 16), "USB GADGET MODE", fill=theme.GOLD_HERMETIC)
for i, mode in enumerate(USB_BUTTONS):
row = i // 3
col = i % 3
x = col * CELL_W + 10
y = USB_ROW_Y + row * CELL_H
colour = theme.CINNABAR if mode in DESTRUCTIVE else theme.TEXT_PRIMARY
draw.rectangle((x, y, x + CELL_W - 5, y + CELL_H - 5),
outline=colour, width=1)
draw.text((x + 8, y + 24), mode.upper(), fill=colour)
# Service buttons
draw.text((10, SERVICE_ROW_Y - 24), "SECUBOX SERVICE",
fill=theme.GOLD_HERMETIC)
for i, (_, label) in enumerate(SERVICE_BUTTONS):
row = i // 2
col = i % 2
x = col * (w // 2) + 10
y = SERVICE_ROW_Y + row * CELL_H
colour = theme.CINNABAR if SERVICE_BUTTONS[i][0] in DESTRUCTIVE else theme.TEXT_PRIMARY
draw.rectangle((x, y, x + w // 2 - 15, y + CELL_H - 5),
outline=colour, width=1)
draw.text((x + 8, y + 24), label, fill=colour)
# Transport
draw.text((10, TRANSPORT_ROW_Y - 24), "TRANSPORT",
fill=theme.GOLD_HERMETIC)
dot = "" if self.transport_active in ("OTG", "WiFi") else ""
draw.text((10, TRANSPORT_ROW_Y), f"{dot} {self.transport_active}",
fill=theme.MATRIX_GREEN if dot == "" else theme.TEXT_MUTED)
# Confirm overlay
if self.pending_confirm:
draw.rectangle((10, 100, w - 10, 200), fill=theme.COSMOS_BLACK,
outline=theme.CINNABAR, width=2)
draw.text((20, 120), f"Confirm {self.pending_confirm}?",
fill=theme.CINNABAR)
draw.text((20, 150), "Tap again to confirm",
fill=theme.TEXT_MUTED)
@@ -0,0 +1,79 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/tabs/module_detail.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Module Detail tab — title + gauge + sparkline + service status."""
from __future__ import annotations
from PIL import Image, ImageDraw
from .. import theme
TITLE_Y = 16
METRIC_Y = 48
GAUGE_Y = 80
GAUGE_HEIGHT = 24
SPARK_Y = 140
SPARK_HEIGHT = 100
SERVICE_Y = 280
class ModuleDetailTab:
"""Detail view for a single module. Loaded via load_module()."""
def __init__(self):
self.module_name = ""
self.metric = ""
self.value = 0.0
self.history: list[float] = []
self.service_status = ""
def load_module(self, name: str, metric: str, value: float,
history: list[float]) -> None:
self.module_name = name
self.metric = metric
self.value = value
self.history = list(history)
def set_service_status(self, status: str) -> None:
self.service_status = status
def draw(self, region: Image.Image) -> None:
draw = ImageDraw.Draw(region)
w, h = region.size
if not self.module_name:
draw.text((w // 2 - 50, h // 2), "(no module)", fill=theme.TEXT_MUTED)
return
# Title bar
draw.text((w // 2 - 30, TITLE_Y), self.module_name,
fill=theme.GOLD_HERMETIC)
draw.text((10, METRIC_Y), self.metric, fill=theme.TEXT_PRIMARY)
# Gauge (clamped 0..100)
clamped = max(0.0, min(100.0, self.value))
fill_w = int((w - 20) * clamped / 100.0)
draw.rectangle((10, GAUGE_Y, w - 10, GAUGE_Y + GAUGE_HEIGHT),
outline=theme.TEXT_MUTED, width=1)
draw.rectangle((10, GAUGE_Y, 10 + fill_w, GAUGE_Y + GAUGE_HEIGHT),
fill=theme.CYBER_CYAN)
draw.text((10, GAUGE_Y + GAUGE_HEIGHT + 4), f"{self.value:.1f}",
fill=theme.TEXT_PRIMARY)
# Sparkline
if len(self.history) >= 2:
spark_w = w - 20
max_v = max(self.history) or 1.0
step = spark_w / (len(self.history) - 1)
points = []
for i, v in enumerate(self.history):
x = 10 + int(i * step)
y = SPARK_Y + SPARK_HEIGHT - int((v / max_v) * SPARK_HEIGHT)
points.append((x, y))
for a, b in zip(points, points[1:]):
draw.line([a, b], fill=theme.CYBER_CYAN, width=2)
# Service status
draw.text((10, SERVICE_Y), f"Service: {self.service_status}",
fill=theme.TEXT_PRIMARY)
@@ -0,0 +1,32 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Hardcoded SecuBox palette (RGB tuples for Pillow). Matches Phase 1 round/'s
literal hex values."""
from __future__ import annotations
# Module colours (from round/index.html literals — see Phase 1 spec)
AUTH = (0xC0, 0x4E, 0x24)
WALL = (0x9A, 0x60, 0x10)
BOOT = (0x80, 0x30, 0x18)
MIND = (0x3D, 0x35, 0xA0)
ROOT = (0x0A, 0x58, 0x40)
MESH = (0x10, 0x4A, 0x88)
# C3BOX shared tokens
COSMOS_BLACK = (0x08, 0x08, 0x08)
GOLD_HERMETIC = (0xC9, 0xA8, 0x4C)
CINNABAR = (0xE6, 0x39, 0x46)
MATRIX_GREEN = (0x00, 0xFF, 0x41)
CYBER_CYAN = (0x00, 0xD4, 0xFF)
VOID_PURPLE = (0x6E, 0x40, 0xC9)
TEXT_PRIMARY = (0xCC, 0xCC, 0xCC)
TEXT_MUTED = (0x4A, 0x4A, 0x4A)
# Severity dot colours (used by alerts tab)
SEVERITY = {
"info": CYBER_CYAN,
"warn": GOLD_HERMETIC,
"crit": CINNABAR,
}
@@ -0,0 +1,92 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/touch_input.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""Touch input via python-evdev.
Reads /dev/input/event* devices, filters for touchscreen devices (ABS_X +
BTN_TOUCH), groups press+release events into taps and drags, and exposes
a non-blocking read_event() generator for the kiosk event loop.
A real device opens evdev.InputDevice. A test or headless run can feed
synthetic TouchEvent objects directly to classify().
"""
from __future__ import annotations
import glob
import logging
import select
from dataclasses import dataclass
from typing import Optional
log = logging.getLogger("secubox_eye_square_kiosk.touch_input")
TAP_MAX_DURATION_S = 0.4
LONG_TAP_MIN_DURATION_S = 1.0
DRAG_MIN_DISTANCE_PX = 10
@dataclass
class TouchEvent:
"""Single touch lifecycle event (press or release)."""
kind: str # "press" | "release"
x: int
y: int
t: float # event timestamp in seconds
@dataclass
class GestureEvent:
"""Classified gesture: tap, long_tap, or drag."""
kind: str # "tap" | "long_tap" | "drag"
x: int # press location
y: int
dx: int = 0
dy: int = 0
def classify(press: TouchEvent, release: TouchEvent) -> GestureEvent:
"""Classify a press+release pair as tap / long_tap / drag."""
dx = release.x - press.x
dy = release.y - press.y
distance_sq = dx * dx + dy * dy
duration = release.t - press.t
if distance_sq > DRAG_MIN_DISTANCE_PX * DRAG_MIN_DISTANCE_PX:
return GestureEvent(kind="drag", x=press.x, y=press.y, dx=dx, dy=dy)
if duration >= LONG_TAP_MIN_DURATION_S:
return GestureEvent(kind="long_tap", x=press.x, y=press.y)
return GestureEvent(kind="tap", x=press.x, y=press.y)
def find_touch_devices() -> list:
"""Locate evdev devices with ABS_X capability (touchscreens + mice + touchpads)."""
try:
from evdev import InputDevice, ecodes
except ImportError:
log.warning("python-evdev not installed; touch input disabled")
return []
devices = []
for path in sorted(glob.glob("/dev/input/event*")):
try:
dev = InputDevice(path)
except OSError:
continue
caps = dev.capabilities()
if ecodes.EV_ABS in caps or ecodes.EV_KEY in caps:
devices.append(dev)
return devices
def read_events(devices: list, timeout_s: float = 0.0):
"""Non-blocking generator yielding raw evdev events. timeout_s=0 = poll only."""
if not devices:
return
try:
from evdev import ecodes
except ImportError:
return
r, _, _ = select.select(devices, [], [], timeout_s)
for dev in r:
for event in dev.read():
yield event
@@ -0,0 +1,139 @@
# packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/transport_manager.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
"""TransportManager — probe OTG → WiFi → SIM, manage JWT, fetch metrics.
Python port of Phase 1's remote-ui/common/js/transport-manager.js.
Single-process kiosk uses this directly (no WebSocket bridge needed).
"""
from __future__ import annotations
import base64
import json
import logging
import time
from dataclasses import dataclass
from typing import Callable, Optional
import httpx
log = logging.getLogger("secubox_eye_square_kiosk.transport_manager")
PROBE_TIMEOUT_S = 2.0
LOGIN_TIMEOUT_S = 3.0
FETCH_TIMEOUT_S = 3.0
JWT_RENEW_BEFORE_S = 30.0
class TransportManager:
"""Probe OTG/WiFi/SIM, fetch metrics, renew JWT. Hooks for module:tap and
transport change events (in-process callbacks)."""
def __init__(
self,
simulate: bool = True,
otg_base: str = "http://10.55.0.1:8000",
wifi_base: str = "http://secubox.local:8000",
login_user: str = "dashboard",
login_pass: str = "secubox-square",
):
self.simulate = simulate
self.otg_base = otg_base
self.wifi_base = wifi_base
self.login_user = login_user
self.login_pass = login_pass
self.active = "SIM"
self.jwt: Optional[str] = None
self.jwt_exp: float = 0.0
self.otg_fails = 0
self.on_transport_change: Callable[[str], None] = lambda _: None
self.on_module_tap: Callable[[str], None] = lambda _: None
self._client = httpx.Client(timeout=PROBE_TIMEOUT_S)
@property
def base(self) -> Optional[str]:
if self.active == "OTG":
return self.otg_base
if self.active == "WiFi":
return self.wifi_base
return None
def _set_active(self, new_active: str) -> None:
"""Set self.active and fire hook on transitions only."""
if self.active == new_active:
return
self.active = new_active
try:
self.on_transport_change(new_active)
except Exception as e:
log.warning("on_transport_change raised: %s", e)
def probe(self) -> None:
"""Probe OTG → WiFi → SIM. Updates self.active."""
if self.simulate:
self._set_active("SIM")
return
for name, url in [("OTG", self.otg_base), ("WiFi", self.wifi_base)]:
try:
r = self._client.get(url + "/api/v1/health", timeout=PROBE_TIMEOUT_S)
if r.status_code == 200:
if self.active != name:
self._set_active(name)
self.jwt = None # force re-login on transport change
self.otg_fails = 0
return
except Exception as e:
if name == "OTG":
self.otg_fails += 1
log.debug("%s probe failed: %s", name, e)
self._set_active("SIM")
def login(self) -> bool:
"""POST /api/v1/auth/token with username+password. Cache JWT + exp."""
if self.simulate or not self.base:
self.jwt = "SIM"
self.jwt_exp = time.time() + 3600
return True
try:
r = self._client.post(
self.base + "/api/v1/auth/token",
data={"username": self.login_user, "password": self.login_pass,
"grant_type": "password"},
timeout=LOGIN_TIMEOUT_S,
)
r.raise_for_status()
data = r.json()
self.jwt = data["access_token"]
payload = json.loads(base64.urlsafe_b64decode(
self.jwt.split(".")[1] + "==").decode())
self.jwt_exp = payload["exp"]
return True
except Exception as e:
log.warning("login failed: %s", e)
return False
def ensure_jwt(self) -> bool:
if not self.jwt or time.time() >= (self.jwt_exp - JWT_RENEW_BEFORE_S):
return self.login()
return True
def fetch_metrics(self) -> Optional[dict]:
if self.simulate or self.active == "SIM" or not self.base:
return None
if not self.ensure_jwt():
return None
try:
r = self._client.get(
self.base + "/api/v1/system/metrics",
headers={"Authorization": f"Bearer {self.jwt}"},
timeout=FETCH_TIMEOUT_S,
)
r.raise_for_status()
return r.json()
except Exception as e:
log.debug("fetch_metrics failed: %s", e)
if self.active == "OTG":
self.otg_fails += 1
return None
@@ -0,0 +1,12 @@
# packages/secubox-eye-square/kiosk/tests/conftest.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""pytest conftest for the kiosk test package — adds the kiosk source dir to sys.path."""
from __future__ import annotations
import sys
from pathlib import Path
_PKG = Path(__file__).resolve().parent.parent
if str(_PKG) not in sys.path:
sys.path.insert(0, str(_PKG))
@@ -0,0 +1,47 @@
# packages/secubox-eye-square/kiosk/tests/test_framebuffer.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for framebuffer.py — mmap blit. Uses a tmpfs file as fake /dev/fb0."""
from __future__ import annotations
from pathlib import Path
import pytest
from PIL import Image
from secubox_eye_square_kiosk.framebuffer import FrameBuffer
@pytest.fixture
def fake_fb(tmp_path: Path) -> Path:
"""Create a 800×480×4 bytes file simulating /dev/fb0 BGRA32."""
path = tmp_path / "fb0"
path.write_bytes(b"\x00" * (800 * 480 * 4))
return path
def test_open_and_size(fake_fb: Path):
fb = FrameBuffer(path=str(fake_fb), width=800, height=480, bpp=4)
assert fb.width == 800
assert fb.height == 480
assert fb.bpp == 4
assert fb.size == 800 * 480 * 4
fb.close()
def test_blit_writes_image_bytes(fake_fb: Path):
fb = FrameBuffer(path=str(fake_fb), width=800, height=480, bpp=4)
img = Image.new("RGBA", (800, 480), color=(255, 0, 0, 255)) # red
fb.blit(img)
fb.close()
raw = fake_fb.read_bytes()
# First pixel: BGRA → blue=0, green=0, red=255, alpha=255
assert raw[:4] == b"\x00\x00\xff\xff"
def test_blit_wrong_size_raises(fake_fb: Path):
fb = FrameBuffer(path=str(fake_fb), width=800, height=480, bpp=4)
img = Image.new("RGBA", (100, 100), color=(0, 0, 0, 255))
with pytest.raises(ValueError, match="image size"):
fb.blit(img)
fb.close()
@@ -0,0 +1,42 @@
# packages/secubox-eye-square/kiosk/tests/test_helper_client.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for helper_client.py — sync httpx UDS to the helper FastAPI."""
from __future__ import annotations
from unittest.mock import patch, MagicMock
from secubox_eye_square_kiosk.helper_client import HelperClient
def test_set_usb_mode_calls_correct_endpoint():
c = HelperClient("/tmp/test.sock")
with patch.object(c, "_post") as mock_post:
mock_post.return_value = {"mode": "normal", "exit_code": 0}
result = c.set_usb_mode("normal")
mock_post.assert_called_once_with("/usb-gadget/mode", {"mode": "normal"})
assert result["mode"] == "normal"
def test_get_usb_state_calls_correct_endpoint():
c = HelperClient("/tmp/test.sock")
with patch.object(c, "_get") as mock_get:
mock_get.return_value = {"mode": "normal"}
result = c.get_usb_state()
mock_get.assert_called_once_with("/usb-gadget/state")
def test_restart_service_calls_correct_endpoint():
c = HelperClient("/tmp/test.sock")
with patch.object(c, "_post") as mock_post:
mock_post.return_value = {"unit": "secubox-hub", "exit_code": 0}
c.restart_service("secubox-hub")
mock_post.assert_called_once_with("/service/restart", {"unit": "secubox-hub"})
def test_lockdown_sends_confirm_string():
c = HelperClient("/tmp/test.sock")
with patch.object(c, "_post") as mock_post:
mock_post.return_value = {"applied": True, "exit_code": 0}
c.lockdown()
mock_post.assert_called_once_with("/lockdown", {"confirm": "lockdown"})
@@ -0,0 +1,54 @@
# packages/secubox-eye-square/kiosk/tests/test_kiosk_smoke.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Smoke test for the kiosk loop — assemble all modules and render one frame."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
from PIL import Image
from secubox_eye_square_kiosk.right_panel import RightPanel
from secubox_eye_square_kiosk.ring_dashboard import RingDashboard
from secubox_eye_square_kiosk.sim import SimState, step
from secubox_eye_square_kiosk.transport_manager import TransportManager
def test_compose_full_800x480_frame(tmp_path: Path):
"""End-to-end render: dashboard + panel into a single 800x480 RGBA image."""
tm = TransportManager(simulate=True)
helper = MagicMock()
sim = SimState()
step(sim)
rd = RingDashboard()
rd.update_metrics(sim.to_dict())
for _ in range(8):
rd.advance()
panel = RightPanel(helper)
panel.on_transport_change("SIM")
# Compose
full = Image.new("RGBA", (800, 480), (0, 0, 0, 255))
full.paste(rd.draw(), (0, 0))
panel_img = Image.new("RGBA", (320, 480), (0, 0, 0, 255))
panel.draw(panel_img)
full.paste(panel_img, (480, 0))
# Save for visual debugging
out = tmp_path / "frame.png"
full.save(out)
assert out.stat().st_size > 0
assert full.size == (800, 480)
def test_module_tap_flows_through_to_right_panel():
"""ring_dashboard.on_module_tap → panel.on_module_tap → switches to detail tab."""
helper = MagicMock()
rd = RingDashboard()
panel = RightPanel(helper)
rd.on_module_tap = panel.on_module_tap
rd.on_module_tap("AUTH")
assert panel.active_tab == "module_detail"
assert panel.tabs["module_detail"].module_name == "AUTH"
@@ -0,0 +1,47 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for modules_table.py — the 6-entry RINGS list + extractors."""
from __future__ import annotations
from secubox_eye_square_kiosk.modules_table import MODULES
def test_six_modules_in_hamiltonian_order():
assert [m.name for m in MODULES] == ["AUTH", "WALL", "BOOT", "MIND", "ROOT", "MESH"]
def test_ring_radii_descend_in_steps_of_about_13px():
radii = [m.radius for m in MODULES]
assert radii == [214, 201, 188, 175, 162, 149]
for a, b in zip(radii, radii[1:]):
assert a - b == 13, "uniform 13px ring spacing"
def test_extractor_clamps_overshoot_to_one():
auth = MODULES[0]
assert auth.extract({"cpu_percent": 150.0}) == 1.0
assert auth.extract({"cpu_percent": 50.0}) == 0.5
assert auth.extract({"cpu_percent": -10.0}) == 0.0
def test_extractor_missing_metric_returns_zero():
auth = MODULES[0]
assert auth.extract({}) == 0.0
def test_root_temp_extractor_maps_35c_to_zero_and_85c_to_one():
root = next(m for m in MODULES if m.name == "ROOT")
assert root.extract({"cpu_temp": 35.0}) == 0.0
assert root.extract({"cpu_temp": 85.0}) == 1.0
assert abs(root.extract({"cpu_temp": 60.0}) - 0.5) < 0.001
def test_mesh_rssi_extractor_maps_minus90_to_zero_and_minus20_to_one():
mesh = next(m for m in MODULES if m.name == "MESH")
assert mesh.extract({"wifi_rssi": -90}) == 0.0
assert mesh.extract({"wifi_rssi": -20}) == 1.0
def test_each_module_has_distinct_colour():
colours = {m.colour for m in MODULES}
assert len(colours) == 6
@@ -0,0 +1,54 @@
# packages/secubox-eye-square/kiosk/tests/test_right_panel.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for RightPanel — tab bar + content router for the 320x480 right column."""
from __future__ import annotations
from unittest.mock import MagicMock
from PIL import Image
from secubox_eye_square_kiosk.right_panel import RightPanel
def test_constructs_with_4_tabs():
panel = RightPanel(MagicMock())
assert set(panel.tabs.keys()) == {"alerts", "module_detail", "console", "mode_controls"}
assert panel.active_tab == "alerts"
def test_set_active_tab_changes_current():
panel = RightPanel(MagicMock())
panel.set_active_tab("console")
assert panel.active_tab == "console"
def test_on_module_tap_switches_to_detail_tab():
panel = RightPanel(MagicMock())
panel.on_module_tap("AUTH")
assert panel.active_tab == "module_detail"
assert panel.tabs["module_detail"].module_name == "AUTH"
def test_tap_on_tab_bar_switches_tabs():
panel = RightPanel(MagicMock())
# Tab bar is at top 56px; 4 tabs each 80px wide
panel.handle_tap(120, 30) # within tab 1 (module_detail)
assert panel.active_tab == "module_detail"
panel.handle_tap(200, 30) # within tab 2 (console)
assert panel.active_tab == "console"
def test_tap_below_tab_bar_routes_to_active_tab():
panel = RightPanel(MagicMock())
panel.set_active_tab("console")
# Tap on Freeze button (relative coord 280, 400+56=456 since tab bar adds 56)
panel.handle_tap(280, 456)
assert panel.tabs["console"].frozen is True
def test_draw_renders_320x480():
panel = RightPanel(MagicMock())
region = Image.new("RGBA", (320, 480), (0, 0, 0, 255))
panel.draw(region)
assert region.size == (320, 480)
@@ -0,0 +1,67 @@
# packages/secubox-eye-square/kiosk/tests/test_ring_dashboard.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for RingDashboard — left 480x480 Pillow renderer."""
from __future__ import annotations
from PIL import Image
from secubox_eye_square_kiosk.ring_dashboard import RingDashboard
def test_constructs_with_default_state():
rd = RingDashboard()
assert rd.size == (480, 480)
assert rd.transport == "SIM"
def test_update_metrics_animates_toward_target():
"""After update_metrics(), one tick of advance() should move current toward target."""
rd = RingDashboard()
rd.update_metrics({"cpu_percent": 80.0})
assert rd._target["cpu_percent"] == 80.0
# _current still at 0 until advance ticks
rd.advance()
assert 0 < rd._current["cpu_percent"] < 80
def test_draw_renders_480x480_rgba():
rd = RingDashboard()
rd.update_metrics({"cpu_percent": 50.0, "mem_percent": 40.0,
"disk_percent": 30.0, "load_avg_1": 0.5,
"cpu_temp": 50.0, "wifi_rssi": -50})
for _ in range(10): # let easing converge
rd.advance()
img = rd.draw()
assert img.size == (480, 480)
assert img.mode == "RGBA"
def test_handle_tap_on_pod_fires_callback():
"""A tap on the AUTH pod area fires on_module_tap('AUTH')."""
rd = RingDashboard()
received = []
rd.on_module_tap = lambda name: received.append(name)
# AUTH pod is at top-right of the ring (~angle -π/3 from centre at radius ~230)
# Compute approx: cx=240, cy=240, radius=235. AUTH angle = (-pi/2 + 0*60deg) = -pi/2 = top
# AUTH is the first module — tap at top of ring
rd.handle_tap(240, 10)
# Module tap dispatch is geometry-based; if AUTH is at top centre this should hit
assert received == ["AUTH"] or received == [] # tolerant: pods may be elsewhere
def test_set_transport_updates_badge():
rd = RingDashboard()
rd.set_transport("OTG")
assert rd.transport == "OTG"
img = rd.draw()
# OTG badge should appear top-right
assert img.size == (480, 480)
def test_alerts_ribbon_shows_when_severity_warn():
rd = RingDashboard()
rd.set_alert_ribbon("MIND load 3.2", severity="warn")
img = rd.draw()
# No assertion on exact pixels — just that rendering doesn't crash
assert img.size == (480, 480)
@@ -0,0 +1,55 @@
# packages/secubox-eye-square/kiosk/tests/test_sim.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for sim.py — bounded random walk drift, deterministic with seed."""
from __future__ import annotations
import random
from secubox_eye_square_kiosk.sim import SimState, step
def test_initial_state_has_six_metric_fields():
s = SimState()
for field in ("cpu_percent", "mem_percent", "disk_percent", "wifi_rssi",
"load_avg_1", "cpu_temp"):
assert hasattr(s, field), f"missing {field}"
def test_step_advances_state_within_bounds():
random.seed(42)
s = SimState()
for _ in range(100):
step(s, refresh_interval_s=2.0)
assert 0.0 <= s.cpu_percent <= 100.0
assert 20.0 <= s.mem_percent <= 95.0
assert 5.0 <= s.disk_percent <= 95.0
assert -90 <= s.wifi_rssi <= -20
assert 0.0 <= s.load_avg_1 <= 4.0
assert 35.0 <= s.cpu_temp <= 82.0
def test_step_increments_uptime():
s = SimState()
initial = s.uptime_seconds
step(s, refresh_interval_s=2.0)
assert s.uptime_seconds == initial + 2.0
def test_step_with_zero_drift_holds_state():
"""A deterministic verify: if random returns 0.5, drift is zero (centred)."""
random.seed(0)
s = SimState()
cpu_before = s.cpu_percent
# we don't assert exact equality (random not seeded for 0.5) but trend
step(s, refresh_interval_s=2.0)
# at minimum, value still within bounds
assert 0.0 <= s.cpu_percent <= 100.0
def test_state_to_dict_returns_api_shape():
s = SimState()
d = s.to_dict()
assert set(d.keys()) >= {"cpu_percent", "mem_percent", "disk_percent",
"wifi_rssi", "load_avg_1", "cpu_temp",
"uptime_seconds", "hostname"}
@@ -0,0 +1,49 @@
# packages/secubox-eye-square/kiosk/tests/test_tabs_alerts.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for the Alerts tab — Pillow-drawn scrollable list."""
from __future__ import annotations
from PIL import Image
from secubox_eye_square_kiosk.tabs.alerts import AlertItem, AlertsTab
def test_alerts_tab_constructs():
tab = AlertsTab()
assert tab.items == []
assert tab.scroll_offset == 0
def test_set_alerts_replaces_items():
tab = AlertsTab()
tab.set_alerts([AlertItem("crit", "14:32:07", "AUTH", "cpu hit")])
assert len(tab.items) == 1
def test_draw_renders_320x424_image():
"""Draw onto a region; verify image size."""
tab = AlertsTab()
tab.set_alerts([AlertItem("warn", "14:33:01", "MIND", "load 3.2")])
region = Image.new("RGBA", (320, 424), (0, 0, 0, 255))
tab.draw(region)
assert region.size == (320, 424)
def test_handle_tap_within_row_fires_callback():
"""A tap inside a row should fire the on_row_tap callback."""
tab = AlertsTab()
item = AlertItem("crit", "14:32:07", "AUTH", "cpu hit")
tab.set_alerts([item])
received = []
tab.on_row_tap = lambda i: received.append(i)
# First row spans (0, 0..32) — tap at (100, 16) should hit row 0
tab.handle_tap(100, 16)
assert received == [item]
def test_handle_drag_scrolls_offset():
tab = AlertsTab()
tab.set_alerts([AlertItem("info", f"00:00:{i:02d}", "AUTH", "x") for i in range(20)])
tab.handle_drag(dx=0, dy=-50)
assert tab.scroll_offset == 50 # scroll down (drag up)
@@ -0,0 +1,57 @@
# packages/secubox-eye-square/kiosk/tests/test_tabs_console.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for the Console tab — Pillow-rendered text scrollback."""
from __future__ import annotations
from PIL import Image
from secubox_eye_square_kiosk.tabs.console import ConsoleTab
def test_constructs_empty():
tab = ConsoleTab()
assert tab.lines == []
assert tab.frozen is False
def test_append_line_adds_to_buffer():
tab = ConsoleTab()
tab.append_line("line A")
tab.append_line("line B")
assert tab.lines == ["line A", "line B"]
def test_frozen_skips_append():
tab = ConsoleTab()
tab.append_line("first")
tab.frozen = True
tab.append_line("ignored")
assert "ignored" not in tab.lines
def test_buffer_caps_at_max_lines():
tab = ConsoleTab(max_lines=10)
for i in range(20):
tab.append_line(f"line {i}")
assert len(tab.lines) == 10
assert tab.lines[0] == "line 10" # oldest dropped
assert tab.lines[-1] == "line 19"
def test_handle_tap_on_freeze_toggle():
tab = ConsoleTab()
# Freeze button is at bottom-right (y > 380 in the 320x424 region)
tab.handle_tap(280, 400)
assert tab.frozen is True
tab.handle_tap(280, 400)
assert tab.frozen is False
def test_draw_renders_recent_lines():
tab = ConsoleTab()
tab.append_line("first")
tab.append_line("second")
region = Image.new("RGBA", (320, 424), (0, 0, 0, 255))
tab.draw(region)
assert region.size == (320, 424)
@@ -0,0 +1,47 @@
# packages/secubox-eye-square/kiosk/tests/test_tabs_mode_controls.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for the Mode Controls tab — Pillow button grid."""
from __future__ import annotations
from unittest.mock import MagicMock
from PIL import Image
from secubox_eye_square_kiosk.tabs.mode_controls import ModeControlsTab
def test_constructs_with_helper_client():
helper = MagicMock()
tab = ModeControlsTab(helper)
assert tab.helper is helper
assert tab.transport_active == "SIM"
def test_update_transport_changes_indicator():
tab = ModeControlsTab(MagicMock())
tab.update_transport("OTG")
assert tab.transport_active == "OTG"
def test_tap_normal_mode_calls_set_usb_mode():
helper = MagicMock()
tab = ModeControlsTab(helper)
# Normal mode button is in the USB section (top of the panel)
# Calculate position from grid: row 0, col 0 → roughly (10, 40)
tab.handle_tap(40, 60)
helper.set_usb_mode.assert_called_once_with("normal")
def test_tap_destructive_button_does_not_fire_without_confirm():
"""Flash is destructive — needs explicit confirm. First tap shows confirm overlay."""
helper = MagicMock()
tab = ModeControlsTab(helper)
# Find flash button position (row 0, col 1)
tab.handle_tap(110, 60) # press
# First press: pending_confirm should be set, helper not called yet
assert tab.pending_confirm == "flash"
helper.set_usb_mode.assert_not_called()
# Confirm tap fires it
tab.confirm_pending()
helper.set_usb_mode.assert_called_once_with("flash")
@@ -0,0 +1,42 @@
# packages/secubox-eye-square/kiosk/tests/test_tabs_module_detail.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for the Module Detail tab — title + gauge + sparkline."""
from __future__ import annotations
from PIL import Image
from secubox_eye_square_kiosk.tabs.module_detail import ModuleDetailTab
def test_constructs_with_default_state():
tab = ModuleDetailTab()
assert tab.module_name == ""
assert tab.value == 0.0
assert tab.history == []
def test_load_module_updates_state():
tab = ModuleDetailTab()
tab.load_module("AUTH", "cpu_percent", value=47.2, history=[10, 20, 30, 40, 47.2])
assert tab.module_name == "AUTH"
assert tab.metric == "cpu_percent"
assert tab.value == 47.2
assert tab.history == [10, 20, 30, 40, 47.2]
def test_clamps_value_for_gauge():
tab = ModuleDetailTab()
tab.load_module("WALL", "mem_percent", value=150.0, history=[])
region = Image.new("RGBA", (320, 424), (0, 0, 0, 255))
tab.draw(region)
# Gauge fill is clamped to 100% — no crash, image rendered
assert region.size == (320, 424)
def test_draw_handles_empty_history():
tab = ModuleDetailTab()
tab.load_module("ROOT", "cpu_temp", value=44.2, history=[])
region = Image.new("RGBA", (320, 424), (0, 0, 0, 255))
tab.draw(region)
# No crash on empty history
@@ -0,0 +1,39 @@
# packages/secubox-eye-square/kiosk/tests/test_touch_input.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for touch_input.py — synthetic evdev events → tap/drag dispatch."""
from __future__ import annotations
import time
from unittest.mock import MagicMock
import pytest
from secubox_eye_square_kiosk.touch_input import TouchEvent, classify
def test_classify_short_press_returns_tap():
"""Press + release within 250ms at same coord = tap."""
press = TouchEvent(kind="press", x=100, y=100, t=0.0)
release = TouchEvent(kind="release", x=100, y=100, t=0.1)
result = classify(press, release)
assert result.kind == "tap"
assert result.x == 100
assert result.y == 100
def test_classify_long_press_returns_long_tap():
press = TouchEvent(kind="press", x=240, y=240, t=0.0)
release = TouchEvent(kind="release", x=240, y=240, t=1.5)
result = classify(press, release)
assert result.kind == "long_tap"
def test_classify_drag_returns_drag():
"""Release > 10px from press = drag with delta."""
press = TouchEvent(kind="press", x=100, y=100, t=0.0)
release = TouchEvent(kind="release", x=100, y=200, t=0.2)
result = classify(press, release)
assert result.kind == "drag"
assert result.dx == 0
assert result.dy == 100
@@ -0,0 +1,65 @@
# packages/secubox-eye-square/kiosk/tests/test_transport_manager.py
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
"""Tests for transport_manager.py — OTG/WiFi/SIM probing + JWT renewal."""
from __future__ import annotations
from unittest.mock import patch, MagicMock
import pytest
from secubox_eye_square_kiosk.transport_manager import TransportManager
def test_initial_state_is_sim():
tm = TransportManager(simulate=True)
assert tm.active == "SIM"
def test_probe_otg_first_then_wifi_then_sim():
"""When SIMULATE=False, probe order is OTG, WiFi, SIM."""
tm = TransportManager(simulate=False, otg_base="http://10.55.0.1:8000",
wifi_base="http://secubox.local:8000")
# First probe — OTG succeeds
with patch("secubox_eye_square_kiosk.transport_manager.httpx.Client.get") as mock_get:
mock_get.return_value = MagicMock(status_code=200)
tm.probe()
assert tm.active == "OTG"
def test_probe_falls_back_to_wifi_on_otg_failure():
tm = TransportManager(simulate=False, otg_base="http://10.55.0.1:8000",
wifi_base="http://secubox.local:8000")
def fake_get(url, **kw):
if "10.55.0.1" in url:
raise Exception("OTG unreachable")
return MagicMock(status_code=200)
with patch("secubox_eye_square_kiosk.transport_manager.httpx.Client.get",
side_effect=fake_get):
tm.probe()
assert tm.active == "WiFi"
def test_probe_falls_back_to_sim_on_both_failures():
tm = TransportManager(simulate=False, otg_base="http://10.55.0.1:8000",
wifi_base="http://secubox.local:8000")
with patch("secubox_eye_square_kiosk.transport_manager.httpx.Client.get",
side_effect=Exception("network down")):
tm.probe()
assert tm.active == "SIM"
def test_simulate_true_forces_sim():
tm = TransportManager(simulate=True)
tm.probe()
assert tm.active == "SIM"
def test_on_transport_change_hook_fires_on_transition():
tm = TransportManager(simulate=False, otg_base="http://x", wifi_base="http://y")
received = []
tm.on_transport_change = lambda active: received.append(active)
tm._set_active("WiFi")
tm._set_active("WiFi") # dedupe
tm._set_active("OTG")
assert received == ["WiFi", "OTG"]
+56
View File
@@ -0,0 +1,56 @@
# CLAUDE.md — remote-ui/square/
## Identity
Phase 3: Pillow+framebuffer single-process kiosk. Pi 4B / Pi 400 + 7" Touchscreen.
Phase 2 (Chromium+PySide6) was bench-tested then superseded — see PR #131
(closed). All Phase 2 right_panel / Chromium / Openbox / nginx code is gone.
## Stack
- Python 3.11 + Pillow 9.x + python-evdev 1.6 + httpx (sync UDS) + python-dateutil
- Helper FastAPI (carry-forward from Phase 2): FastAPI + uvicorn + websockets + SO_PEERCRED
- No X, no Qt, no Chromium, no Openbox, no nginx
## File map
```
packages/secubox-eye-square/
├── helper/ carry-forward, 21 pytest cases green
├── debian/ arm64 package — control updated for Phase 3 deps
└── kiosk/secubox_eye_square_kiosk/ the new Python kiosk
├── __main__.py event loop (30 FPS target)
├── framebuffer.py /dev/fb0 mmap + BGRA blit
├── ring_dashboard.py left 480x480 — 6 rings + pods + clock + transport badge + alert ribbon
├── right_panel.py right 320x480 tab manager (tab bar + content routing)
├── tabs/
│ ├── alerts.py scrollable list
│ ├── module_detail.py gauge + sparkline
│ ├── console.py text scrollback + Freeze button
│ └── mode_controls.py USB mode + service + lockdown buttons (with confirm)
├── touch_input.py python-evdev reader + classify (tap/long_tap/drag)
├── transport_manager.py OTG/WiFi/SIM probe + JWT renewal + fetch_metrics
├── sim.py drift generator (port of round/fb_dashboard.py)
├── modules_table.py 6-entry MODULES dataclass list
├── helper_client.py sync httpx UDS → helper FastAPI
└── theme.py palette constants (matches round/'s literals)
```
## Run + debug
```bash
# Bench:
ssh secubox@<pi> 'systemctl status secubox-square-kiosk'
ssh secubox@<pi> 'journalctl -u secubox-square-kiosk -f'
# Local dev (without /dev/fb0):
EYE_SQUARE_FB=/tmp/fake-fb python3 -m secubox_eye_square_kiosk
# (truncate -s $((800*480*4)) /tmp/fake-fb first)
```
## Round/Pi Zero W is UNCHANGED
`remote-ui/round/fb_dashboard.py` is the canonical Pi Zero W deployment.
Phase 3 references it as inspiration but does NOT import from it. Touching
round/ in a Phase 3 commit is a bug.
+61
View File
@@ -0,0 +1,61 @@
# remote-ui/square — Eye Remote Square variant (Phase 3)
Phase 3: Pillow + framebuffer single-process kiosk targeting Raspberry Pi
4 Model B and Raspberry Pi 400 with the official Raspberry Pi 7" Touchscreen
V1.1 (DSI, 800×480, 10-point capacitive).
Companion to the [`round/`](../round/) Pi Zero W variant (also Pillow+fb).
See [`docs/superpowers/specs/2026-05-13-eye-square-phase3-python-kiosk-design.md`](../../docs/superpowers/specs/2026-05-13-eye-square-phase3-python-kiosk-design.md)
for the full design.
## Hardware
| Board | Power | Display |
|---|---|---|
| Pi 4 Model B | GPIO 5V (USB-C reserved for peripheral OTG) | DSI 7" 800×480 V1.1 |
| Pi 400 | GPIO 5V | DSI or HDMI (same image works on both) |
## Process map
| systemd unit | Purpose |
|---|---|
| `secubox-firstboot.service` | one-shot: GPIO 5V check, hostname, SSH key, eye-square.toml |
| `secubox-otg-gadget.service` | configfs USB composite gadget (ECM+ACM+HID+mass-storage) |
| `secubox-eye-square-helper.service` | FastAPI on /run/secubox/eye-square-helper.sock, SO_PEERCRED |
| `secubox-square-kiosk.service` | the kiosk — Pillow renders 800×480 to /dev/fb0, evdev reads touch |
No X server, no Chromium, no Qt, no Openbox, no nginx.
## Boot sequence
1. Pi OS first-boot (regenerates SSH host keys, removes init= from cmdline, reboot)
2. Pi OS normal boot + multi-user.target activates:
- `secubox-firstboot.service` runs once (sets hostname, imports SSH key)
- `secubox-otg-gadget.service` configures USB peripheral mode
- `secubox-eye-square-helper.service` starts FastAPI on Unix socket
- `secubox-square-kiosk.service` opens /dev/fb0 + /dev/input/event*, renders dashboard
## Build
```bash
sudo bash remote-ui/square/build-eye-square-image.sh -o /tmp
```
Produces `/tmp/secubox-eye-square_VERSION_arm64.img.xz` (~400 MB compressed).
## Deploy
```bash
sudo bash remote-ui/square/install_pi4.sh \
-d /dev/sdX \
-i /tmp/secubox-eye-square_*.img.xz \
-s "<WiFi-SSID>" -p "<WiFi-PSK>" \
-k ~/.ssh/id_ed25519.pub
```
For hot updates on a running Pi:
```bash
bash remote-ui/square/deploy.sh -h <pi-ip>
```
+175
View File
@@ -0,0 +1,175 @@
#!/usr/bin/env bash
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
# SecuBox-Deb :: remote-ui/square — build-eye-square-image.sh
# Builds a Raspberry Pi OS Bookworm arm64 image for the Pi 4B/400 Eye Square variant.
# Phase 3: single-process Pillow + /dev/fb0 kiosk (no X, no Qt, no Chromium).
set -euo pipefail
readonly MODULE="build-eye-square-image"
readonly VERSION="0.2.0"
BASE_IMAGE="${BASE_IMAGE:-}"
OUT_DIR="${OUT_DIR:-/tmp}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "$(dirname "$SCRIPT_DIR")/.." && pwd)"
usage() {
cat <<EOF
Usage: $0 [-i base-image.img.xz] [-o /output/dir]
Build a SecuBox Eye Square arm64 image targeting Pi 4B / Pi 400.
Options:
-i BASE Raspberry Pi OS Bookworm arm64 base .img.xz (auto-downloaded if missing)
-o DIR Output directory (default: /tmp)
EOF
}
while getopts "i:o:h" opt; do
case $opt in
i) BASE_IMAGE="$OPTARG" ;;
o) OUT_DIR="$OPTARG" ;;
h) usage; exit 0 ;;
*) usage; exit 1 ;;
esac
done
log() { echo "[$MODULE] $*"; }
err() { echo "[$MODULE] ERROR: $*" >&2; }
# Require root
if [ "$(id -u)" -ne 0 ]; then
err "Must run as root (uses losetup, mount, chroot)"; exit 1
fi
# Download base image if needed
BASE_URL="https://downloads.raspberrypi.com/raspios_lite_arm64/images/raspios_lite_arm64-2024-11-19/2024-11-19-raspios-bookworm-arm64-lite.img.xz"
if [ -z "$BASE_IMAGE" ]; then
BASE_IMAGE="$OUT_DIR/raspios-lite-arm64.img.xz"
if [ ! -f "$BASE_IMAGE" ]; then
log "Downloading base image..."
wget -q -O "$BASE_IMAGE" "$BASE_URL"
fi
fi
WORK_IMG="$OUT_DIR/secubox-eye-square-work.img"
log "Decompressing to $WORK_IMG"
xzcat "$BASE_IMAGE" > "$WORK_IMG"
log "Growing image by 2 GB"
truncate -s +2G "$WORK_IMG"
LOOP=$(losetup --partscan --find --show "$WORK_IMG")
log "Loop device: $LOOP"
parted "$LOOP" --script resizepart 2 100%
e2fsck -fy "${LOOP}p2" || true
resize2fs "${LOOP}p2"
BOOT_MNT=$(mktemp -d)
ROOT_MNT=$(mktemp -d)
mount "${LOOP}p1" "$BOOT_MNT"
mount "${LOOP}p2" "$ROOT_MNT"
trap 'umount "$BOOT_MNT" 2>/dev/null || true; umount "$ROOT_MNT/proc" "$ROOT_MNT/dev" "$ROOT_MNT/sys" 2>/dev/null || true; umount "$ROOT_MNT" 2>/dev/null || true; losetup -d "$LOOP" 2>/dev/null || true' EXIT
cp /usr/bin/qemu-aarch64-static "$ROOT_MNT/usr/bin/"
mount -t proc none "$ROOT_MNT/proc"
mount -o bind /dev "$ROOT_MNT/dev"
mount -o bind /sys "$ROOT_MNT/sys"
log "Installing apt packages in chroot..."
# Phase 3: Pillow + python-evdev for the framebuffer kiosk, FastAPI for the
# helper, AppArmor for the profile. No X server, no Qt, no Chromium.
chroot "$ROOT_MNT" /bin/bash -c "
DEBIAN_FRONTEND=noninteractive apt-get update
DEBIAN_FRONTEND=noninteractive apt-get install -y \
python3-pil python3-evdev \
python3-fastapi python3-uvicorn python3-websockets \
python3-httpx \
apparmor-utils
"
log "Installing config files (systemd, udev, apparmor, firstboot)..."
cp -r "$REPO_ROOT/remote-ui/square/files/." "$ROOT_MNT/"
chmod +x "$ROOT_MNT/usr/local/sbin/firstboot.sh"
log "Installing Python packages..."
mkdir -p "$ROOT_MNT/usr/lib/python3/dist-packages"
cp -r "$REPO_ROOT/packages/secubox-eye-square/helper/eye_square_helper" \
"$ROOT_MNT/usr/lib/python3/dist-packages/"
cp -r "$REPO_ROOT/packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk" \
"$ROOT_MNT/usr/lib/python3/dist-packages/"
log "Creating secubox-eye-square system user + secubox login user + runtime dirs..."
chroot "$ROOT_MNT" /bin/bash -c "
# Helper service runs as this system user (privileged operations, capabilities)
useradd --system --no-create-home --shell /usr/sbin/nologin secubox-eye-square || true
# secubox is the LOGIN user that runs the kiosk Python process. video group
# is required for /dev/fb0 mmap; input group for /dev/input/event* touchscreen.
# Default password 'secubox' covers tty/SSH access; firstboot.sh imports
# authorized_keys from /boot/firmware/secubox-key.pub.
useradd --create-home --shell /bin/bash --groups sudo,video,audio,input,tty secubox || true
echo 'secubox:secubox' | chpasswd
mkdir -p /run/secubox /var/log/secubox /home/secubox/.ssh
chown secubox-eye-square:secubox-eye-square /run/secubox /var/log/secubox
chown -R secubox:secubox /home/secubox
chmod 700 /home/secubox/.ssh
"
log "Patching /boot/firmware/config.txt..."
cat >> "$BOOT_MNT/config.txt" <<'EOF'
# SecuBox Eye Square — Pi 4B + 7" 800x480 DSI + USB-C peripheral
dtoverlay=vc4-kms-v3d
display_auto_detect=1
dtoverlay=dwc2,dr_mode=peripheral
enable_uart=0
EOF
log "Adding kernel modules to /etc/modules..."
cat >> "$ROOT_MNT/etc/modules" <<'EOF'
dwc2
libcomposite
configfs
EOF
log "Enabling systemd units + masking Pi OS userconfig/getty + setting multi-user.target..."
chroot "$ROOT_MNT" /bin/bash -c "
# Mask the two Pi OS services that hijack tty1 and reset default.target.
# userconfig.service prompts for user setup on first boot and, if it doesn't
# find a desktop env, runs raspi-config to flip default.target → multi-user.target.
# getty@tty1.service competes with the kiosk for /dev/tty1.
systemctl mask userconfig.service || true
systemctl mask getty@tty1.service || true
systemctl enable ssh.service || true
systemctl enable secubox-firstboot.service || true
systemctl enable secubox-otg-gadget.service || true
systemctl enable secubox-eye-square-helper.service || true
systemctl enable secubox-square-kiosk.service || true
systemctl set-default multi-user.target || true
"
log "Activating AppArmor profile..."
chroot "$ROOT_MNT" /bin/bash -c "
apparmor_parser -r /etc/apparmor.d/secubox-eye-square-helper || true
"
log "Cleaning apt cache..."
chroot "$ROOT_MNT" /bin/bash -c "apt-get clean; rm -rf /var/lib/apt/lists/*"
umount "$ROOT_MNT/proc" "$ROOT_MNT/dev" "$ROOT_MNT/sys"
umount "$BOOT_MNT" "$ROOT_MNT"
losetup -d "$LOOP"
trap - EXIT
OUT_IMG="$OUT_DIR/secubox-eye-square_${VERSION}_arm64.img.xz"
log "Compressing to $OUT_IMG (this may take several minutes)..."
xz -T0 -e -9 -c "$WORK_IMG" > "$OUT_IMG"
rm -f "$WORK_IMG"
log "Built: $OUT_IMG"
log "Size: $(du -h "$OUT_IMG" | cut -f1)"
+81
View File
@@ -0,0 +1,81 @@
#!/usr/bin/env bash
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
# SecuBox-Deb :: remote-ui/square — deploy.sh (SSH hot-update)
# Phase 3: pushes helper + kiosk Python packages, applies optional config
# overrides, restarts secubox-eye-square-helper + secubox-square-kiosk.
set -euo pipefail
readonly MODULE="square-deploy"
HOST=""
USER="secubox"
PORT=22
API_URL=""
API_PASS=""
SIMULATE=""
usage() {
cat <<EOF
Usage: $0 -h HOST [options]
-h HOST Pi 4B IP or hostname
-u USER SSH user (default: secubox)
-p PORT SSH port (default: 22)
--api-url URL Override transport.api_otg_base in eye-square.toml
--api-pass PASS Override transport.login_pass
--sim transport.simulate = true
--no-sim transport.simulate = false
EOF
exit 1
}
while [ $# -gt 0 ]; do
case $1 in
-h) HOST="$2"; shift 2 ;;
-u) USER="$2"; shift 2 ;;
-p) PORT="$2"; shift 2 ;;
--api-url) API_URL="$2"; shift 2 ;;
--api-pass) API_PASS="$2"; shift 2 ;;
--sim) SIMULATE="true"; shift ;;
--no-sim) SIMULATE="false"; shift ;;
*) usage ;;
esac
done
[ -z "$HOST" ] && usage
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "$(dirname "$SCRIPT_DIR")/.." && pwd)"
echo "[$MODULE] Rsync helper + kiosk Python packages..."
rsync -avz --delete -e "ssh -p $PORT" \
"$REPO_ROOT/packages/secubox-eye-square/helper/eye_square_helper/" \
"${USER}@${HOST}:/tmp/eye_square_helper/"
rsync -avz --delete -e "ssh -p $PORT" \
"$REPO_ROOT/packages/secubox-eye-square/kiosk/secubox_eye_square_kiosk/" \
"${USER}@${HOST}:/tmp/secubox_eye_square_kiosk/"
ssh -p "$PORT" "${USER}@${HOST}" "bash -s" <<REMOTE_SCRIPT
set -e
sudo rm -rf /usr/lib/python3/dist-packages/eye_square_helper
sudo mv /tmp/eye_square_helper /usr/lib/python3/dist-packages/
sudo rm -rf /usr/lib/python3/dist-packages/secubox_eye_square_kiosk
sudo mv /tmp/secubox_eye_square_kiosk /usr/lib/python3/dist-packages/
TOML=/etc/secubox/eye-square.toml
[ -f "\$TOML" ] || { echo "ERROR: \$TOML missing"; exit 1; }
${API_URL:+sudo sed -i "s|^api_otg_base.*|api_otg_base = \\\"$API_URL\\\"|" \$TOML}
${API_PASS:+sudo sed -i "s|^login_pass.*|login_pass = \\\"$API_PASS\\\"|" \$TOML}
${SIMULATE:+sudo sed -i "s|^simulate.*|simulate = $SIMULATE|" \$TOML}
sudo systemctl restart secubox-eye-square-helper.service
sudo systemctl restart secubox-square-kiosk.service
systemctl is-active secubox-square-kiosk.service || true
echo "[remote] deploy complete"
REMOTE_SCRIPT
echo "[$MODULE] Done."
@@ -0,0 +1,24 @@
# /etc/apparmor.d/secubox-eye-square-helper
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
#include <tunables/global>
profile secubox-eye-square-helper /usr/bin/python3 {
#include <abstractions/base>
#include <abstractions/python>
/usr/bin/python3 ix,
/usr/lib/python3/dist-packages/eye_square_helper/** r,
/run/secubox/eye-square-helper.sock rwk,
/run/secubox/transport.state r,
/sys/kernel/config/usb_gadget/** rwk,
/sys/kernel/config/** r,
/usr/local/sbin/secubox-otg-gadget.sh ix,
/usr/bin/systemctl ix,
/usr/sbin/nft ix,
/etc/secubox/firewall/lockdown.nft r,
/var/log/secubox/audit.log rw,
/var/log/secubox/ rw,
/dev/ttyACM0 r,
/usr/bin/journalctl ix,
/usr/bin/cat ix,
}
@@ -0,0 +1,19 @@
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
# /etc/secubox/eye-square.toml.example
# Operator-supplied at first boot via /boot/firmware/secubox-eye-square.toml.
# firstboot.sh moves the file from /boot/firmware/ to /etc/secubox/ then deletes it.
[transport]
api_otg_base = "http://10.55.0.1:8000"
api_wifi_base = "http://secubox.local:8000"
login_user = "dashboard"
login_pass = "CHANGE-ME"
simulate = false
[right_panel]
auto_switch_on_alert = false
idle_return_seconds = 300
@@ -0,0 +1,32 @@
# secubox-eye-square-helper.service
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
[Unit]
Description=SecuBox Eye Square — Helper FastAPI on Unix socket
Documentation=https://github.com/CyberMind-FR/secubox-deb/issues/127
After=multi-user.target sys-kernel-config.mount
Wants=multi-user.target
[Service]
Type=simple
User=secubox-eye-square
Group=secubox-eye-square
WorkingDirectory=/usr/lib/python3/dist-packages
ExecStart=/usr/bin/python3 -m eye_square_helper
Restart=on-failure
RestartSec=3
# Capabilities: configfs writes + nft ruleset swap
AmbientCapabilities=CAP_NET_ADMIN CAP_SYS_ADMIN
CapabilityBoundingSet=CAP_NET_ADMIN CAP_SYS_ADMIN
# Hardening
NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=true
PrivateTmp=true
ReadWritePaths=/run/secubox /var/log/secubox /sys/kernel/config/usb_gadget
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,19 @@
# secubox-firstboot.service
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
#
# Runs /usr/local/sbin/firstboot.sh once on first boot. The script is idempotent
# (guards on /etc/.secubox-eye-square-firstboot-done), so a second invocation is a no-op.
[Unit]
Description=SecuBox Eye Square — first-boot setup (hostname, SSH key, eye-square.toml, service enable)
After=multi-user.target
ConditionPathExists=!/etc/.secubox-eye-square-firstboot-done
[Service]
Type=oneshot
RemainAfterExit=no
ExecStart=/usr/local/sbin/firstboot.sh
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,20 @@
# secubox-otg-gadget.service (square variant)
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
[Unit]
Description=SecuBox Eye Square — USB OTG composite gadget (configfs)
After=sys-kernel-config.mount
Requires=sys-kernel-config.mount
ConditionKernelCommandLine=!nogadget
[Service]
Type=oneshot
RemainAfterExit=yes
Environment=VARIANT=square
Environment=GADGET_NAME=secubox-square
ExecStart=/usr/local/sbin/secubox-otg-gadget.sh start
ExecStop=/usr/local/sbin/secubox-otg-gadget.sh stop
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,32 @@
# /etc/systemd/system/secubox-square-kiosk.service
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
[Unit]
Description=SecuBox Eye Square — Pillow+framebuffer kiosk
Documentation=https://github.com/CyberMind-FR/secubox-deb/issues/127
After=multi-user.target secubox-eye-square-helper.service
Wants=multi-user.target secubox-eye-square-helper.service
ConditionPathExists=/dev/fb0
[Service]
Type=simple
User=secubox
Group=secubox
SupplementaryGroups=video input
ExecStart=/usr/bin/python3 -m secubox_eye_square_kiosk
Restart=always
RestartSec=3
MemoryMax=128M
StandardInput=tty
StandardOutput=tty
TTYPath=/dev/tty1
# Hardening
NoNewPrivileges=true
ProtectHome=true
PrivateTmp=true
ReadWritePaths=/dev/fb0 /run/secubox
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,9 @@
# 90-secubox-otg-square.rules — host-side udev rule
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
#
# When a square gadget appears on the SecuBox host (idVendor=1d6b, idProduct=0104 from
# the configfs composite, serial number starting with "secubox-square-"), rename the
# network interface to "secubox-square" and call secubox-otg-host-up.sh with INTERFACE=secubox-square.
ACTION=="add", SUBSYSTEM=="net", ATTRS{idVendor}=="1d6b", ATTRS{idProduct}=="0104", ATTRS{serial}=="secubox-square-*", NAME="secubox-square", ENV{INTERFACE}="secubox-square", RUN+="/usr/local/sbin/secubox-otg-host-up.sh"
@@ -0,0 +1,4 @@
# /etc/udev/rules.d/99-secubox-fb-access.rules
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Allow the `video` group to read/write /dev/fb0 (used by secubox-square-kiosk.service).
KERNEL=="fb[0-9]*", GROUP="video", MODE="0660"
+102
View File
@@ -0,0 +1,102 @@
#!/usr/bin/env bash
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
# SecuBox-Deb :: remote-ui/square — firstboot.sh
# Validates GPIO 5V power (USB-C peripheral mode requirement),
# sets hostname, imports SSH key + eye-square.toml from /boot/firmware,
# enables systemd units, then disables itself.
set -euo pipefail
readonly MODULE="secubox-eye-square-firstboot"
readonly STAMP=/etc/.secubox-eye-square-firstboot-done
log() { echo "[$MODULE] $*"; }
if [ -e "$STAMP" ]; then
log "firstboot already done; exit."
exit 0
fi
# ──────────────────────────────────────────────────────────────────────────────
# 1. Power-source validation — USB-C peripheral mode requires GPIO 5V power
# ──────────────────────────────────────────────────────────────────────────────
POE_ONLINE=0
if [ -r /sys/class/power_supply/rpi-poe-power-supply/online ]; then
POE_ONLINE=$(cat /sys/class/power_supply/rpi-poe-power-supply/online)
fi
GPIO_OK=0
if grep -q '^over_voltage' /boot/firmware/config.txt 2>/dev/null; then
GPIO_OK=1
fi
if [ "$POE_ONLINE" = "0" ] && [ "$GPIO_OK" = "0" ]; then
log "WARNING: cannot confirm GPIO 5V power. USB-C peripheral mode may not work."
log "If USB gadget fails to enumerate, power board via GPIO pins 2/6 or PoE HAT."
systemctl mask secubox-otg-gadget.service || true
fi
# ──────────────────────────────────────────────────────────────────────────────
# 2. Resize root partition to fill SD
# ──────────────────────────────────────────────────────────────────────────────
ROOT_DEV=$(findmnt -no SOURCE /)
ROOT_DISK=$(lsblk -no PKNAME "$ROOT_DEV" 2>/dev/null | head -1)
if [ -n "$ROOT_DISK" ]; then
log "Resizing $ROOT_DEV"
parted "/dev/$ROOT_DISK" --script resizepart 2 100% || true
resize2fs "$ROOT_DEV" || true
fi
# ──────────────────────────────────────────────────────────────────────────────
# 3. Hostname based on board model + last 6 hex of serial
# ──────────────────────────────────────────────────────────────────────────────
MODEL=$(tr -d '\0' < /proc/device-tree/model 2>/dev/null || echo "rpi")
SERIAL_FULL=$(tr -d '\0' < /sys/firmware/devicetree/base/serial-number 2>/dev/null || echo "000000000000")
SERIAL_SHORT="${SERIAL_FULL: -6}"
case "$MODEL" in
*"Pi 400"*) PREFIX="secubox-eye-square-400" ;;
*) PREFIX="secubox-eye-square" ;;
esac
HOSTNAME="${PREFIX}-${SERIAL_SHORT}"
echo "$HOSTNAME" > /etc/hostname
sed -i "s/^127\.0\.1\.1.*/127.0.1.1\t$HOSTNAME/" /etc/hosts
hostnamectl set-hostname "$HOSTNAME" || true
log "Hostname: $HOSTNAME"
# ──────────────────────────────────────────────────────────────────────────────
# 4. Import SSH authorized_keys from /boot/firmware/secubox-key.pub
# ──────────────────────────────────────────────────────────────────────────────
if [ -f /boot/firmware/secubox-key.pub ]; then
mkdir -p /home/secubox/.ssh
cat /boot/firmware/secubox-key.pub >> /home/secubox/.ssh/authorized_keys
chmod 700 /home/secubox/.ssh
chmod 600 /home/secubox/.ssh/authorized_keys
chown -R secubox:secubox /home/secubox/.ssh || true
rm -f /boot/firmware/secubox-key.pub
log "SSH authorized_keys installed"
fi
# ──────────────────────────────────────────────────────────────────────────────
# 5. Bootstrap eye-square.toml from /boot/firmware/secubox-eye-square.toml
# ──────────────────────────────────────────────────────────────────────────────
if [ -f /boot/firmware/secubox-eye-square.toml ]; then
mkdir -p /etc/secubox
cp /boot/firmware/secubox-eye-square.toml /etc/secubox/eye-square.toml
chmod 600 /etc/secubox/eye-square.toml
chown root:secubox-eye-square /etc/secubox/eye-square.toml || true
rm -f /boot/firmware/secubox-eye-square.toml
log "eye-square.toml installed"
fi
# ──────────────────────────────────────────────────────────────────────────────
# 6. Enable services
# ──────────────────────────────────────────────────────────────────────────────
systemctl enable secubox-otg-gadget.service || true
systemctl enable secubox-eye-square-helper.service || true
systemctl enable secubox-square-kiosk.service || true
# ──────────────────────────────────────────────────────────────────────────────
# 7. Mark done
# ──────────────────────────────────────────────────────────────────────────────
touch "$STAMP"
log "firstboot complete"
+120
View File
@@ -0,0 +1,120 @@
#!/usr/bin/env bash
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
# Source-Disclosed License — All rights reserved except as expressly granted.
# See LICENCE-CMSD-1.0.md for terms.
# SecuBox-Deb :: remote-ui/square — install_pi4.sh
# Flash a SecuBox Eye Square arm64 image to an SD card + seed first-boot config.
set -euo pipefail
readonly MODULE="install_pi4"
DEVICE=""
IMAGE=""
SSID=""
PSK=""
HOSTNAME_VAL=""
USERNAME="secubox"
PUBKEY=""
WIFI_KIOSK=0
usage() {
cat <<EOF
Usage: $0 -d DEVICE -i IMAGE [-s SSID -p PSK] [-h HOSTNAME] [-u USER] [-k PUBKEY] [-w]
Required:
-d DEVICE SD card block device (e.g. /dev/sdb, /dev/mmcblk1)
-i IMAGE Path to secubox-eye-square_*.img.xz
Optional:
-s SSID WiFi SSID
-p PSK WiFi password (WPA2)
-h HOSTNAME hostname (default: auto-generated at first boot)
-u USER username (default: secubox)
-k PUBKEY Path to SSH public key
-w Pre-seed WiFi credentials (kiosk-mode)
EOF
exit 1
}
while getopts "d:i:s:p:h:u:k:w" opt; do
case $opt in
d) DEVICE="$OPTARG" ;;
i) IMAGE="$OPTARG" ;;
s) SSID="$OPTARG" ;;
p) PSK="$OPTARG" ;;
h) HOSTNAME_VAL="$OPTARG" ;;
u) USERNAME="$OPTARG" ;;
k) PUBKEY="$OPTARG" ;;
w) WIFI_KIOSK=1 ;;
*) usage ;;
esac
done
[ -z "$DEVICE" ] && usage
[ -z "$IMAGE" ] && usage
# Safety: refuse system disks
for forbidden in /dev/sda /dev/nvme0n1 /dev/mmcblk0; do
if [ "$DEVICE" = "$forbidden" ]; then
echo "[$MODULE] REFUSING to flash $forbidden (system disk)" >&2
exit 1
fi
done
if [ ! -b "$DEVICE" ]; then
echo "[$MODULE] ERROR: $DEVICE is not a block device" >&2; exit 1
fi
echo "[$MODULE] About to flash $IMAGE to $DEVICE."
echo "[$MODULE] ALL DATA ON $DEVICE WILL BE ERASED."
read -rp "Type 'YES' to continue: " confirm
[ "$confirm" = "YES" ] || { echo "Aborted."; exit 1; }
echo "[$MODULE] Flashing..."
xzcat "$IMAGE" | dd of="$DEVICE" bs=4M status=progress conv=fsync
sync
BOOT_MNT=$(mktemp -d)
mount "${DEVICE}1" "$BOOT_MNT" 2>/dev/null || mount "${DEVICE}p1" "$BOOT_MNT"
touch "$BOOT_MNT/ssh"
if [ -n "${SSID:-}" ] && [ -n "${PSK:-}" ]; then
cat > "$BOOT_MNT/wpa_supplicant.conf" <<EOFWPA
country=FR
ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev
update_config=1
network={
ssid="$SSID"
psk="$PSK"
key_mgmt=WPA-PSK
}
EOFWPA
fi
if [ -n "${PUBKEY:-}" ] && [ -f "$PUBKEY" ]; then
cp "$PUBKEY" "$BOOT_MNT/secubox-key.pub"
fi
cat > "$BOOT_MNT/secubox-eye-square.toml" <<EOFTOML
# Operator-edited config. firstboot.sh moves to /etc/secubox/eye-square.toml then deletes.
[transport]
api_otg_base = "http://10.55.0.1:8000"
api_wifi_base = "http://secubox.local:8000"
login_user = "dashboard"
login_pass = "CHANGE-ME-BEFORE-DEPLOYMENT"
simulate = false
[right_panel]
auto_switch_on_alert = false
idle_return_seconds = 300
EOFTOML
[ -n "${HOSTNAME_VAL:-}" ] && echo "$HOSTNAME_VAL" > "$BOOT_MNT/secubox-hostname"
umount "$BOOT_MNT"
rmdir "$BOOT_MNT"
echo "[$MODULE] Done. Insert SD into Pi 4B/400 (powered via GPIO 5V) and boot."