409 lines
14 KiB
Go
409 lines
14 KiB
Go
// Copyright (c) 2026 Storm Dragon
|
|
// Authors: Storm Dragon, Angus Kola
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
package wayland
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.stormux.org/storm/cthulhu/internal/input"
|
|
"github.com/godbus/dbus/v5"
|
|
)
|
|
|
|
type MonitorSnapshot struct {
|
|
modifiers []uint32
|
|
grabs []keyGrab
|
|
watching, grabbed bool
|
|
client string
|
|
}
|
|
|
|
// FakeKeyboardMonitor verifies the real D-Bus wire contract without pretending
|
|
// to prove physical key suppression by a compositor.
|
|
type FakeKeyboardMonitor struct {
|
|
connection *dbus.Conn
|
|
mutex sync.Mutex
|
|
state MonitorSnapshot
|
|
changes chan MonitorSnapshot
|
|
rejectNext bool
|
|
rejectWatch bool
|
|
}
|
|
|
|
func (monitor *FakeKeyboardMonitor) publish() {
|
|
copy := monitor.state
|
|
copy.modifiers = append([]uint32(nil), copy.modifiers...)
|
|
copy.grabs = append([]keyGrab(nil), copy.grabs...)
|
|
monitor.changes <- copy
|
|
}
|
|
|
|
func (monitor *FakeKeyboardMonitor) WatchKeyboard(sender dbus.Sender) *dbus.Error {
|
|
monitor.mutex.Lock()
|
|
defer monitor.mutex.Unlock()
|
|
if monitor.rejectWatch {
|
|
return dbus.NewError("org.freedesktop.DBus.Error.AccessDenied", []any{"watch denied"})
|
|
}
|
|
monitor.state.client, monitor.state.watching = string(sender), true
|
|
monitor.publish()
|
|
return nil
|
|
}
|
|
func (monitor *FakeKeyboardMonitor) UnwatchKeyboard() *dbus.Error {
|
|
monitor.mutex.Lock()
|
|
defer monitor.mutex.Unlock()
|
|
monitor.state.watching = false
|
|
monitor.publish()
|
|
return nil
|
|
}
|
|
func (monitor *FakeKeyboardMonitor) GrabKeyboard() *dbus.Error {
|
|
monitor.mutex.Lock()
|
|
defer monitor.mutex.Unlock()
|
|
monitor.state.grabbed = true
|
|
monitor.publish()
|
|
return nil
|
|
}
|
|
func (monitor *FakeKeyboardMonitor) UngrabKeyboard() *dbus.Error {
|
|
monitor.mutex.Lock()
|
|
defer monitor.mutex.Unlock()
|
|
monitor.state.grabbed = false
|
|
monitor.publish()
|
|
return nil
|
|
}
|
|
func (monitor *FakeKeyboardMonitor) SetKeyGrabs(modifiers []uint32, grabs []keyGrab) *dbus.Error {
|
|
monitor.mutex.Lock()
|
|
defer monitor.mutex.Unlock()
|
|
if monitor.rejectNext {
|
|
monitor.rejectNext = false
|
|
return dbus.NewError("org.freedesktop.DBus.Error.Failed", []any{"grab update denied"})
|
|
}
|
|
monitor.state.modifiers, monitor.state.grabs = modifiers, grabs
|
|
monitor.publish()
|
|
return nil
|
|
}
|
|
|
|
func start_monitor(t *testing.T) *FakeKeyboardMonitor {
|
|
t.Helper()
|
|
if os.Getenv("CTHULHU_NATIVE_DBUS_TEST") != "1" {
|
|
t.Skip("requires CTHULHU_NATIVE_DBUS_TEST=1 and a private dbus-run-session")
|
|
}
|
|
if os.Getenv("DBUS_SESSION_BUS_ADDRESS") == "" {
|
|
t.Fatal("missing private session bus")
|
|
}
|
|
connection, err := dbus.ConnectSessionBus()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { connection.Close() })
|
|
monitor := &FakeKeyboardMonitor{connection: connection, changes: make(chan MonitorSnapshot, 128)}
|
|
if err := connection.Export(monitor, managerPath, monitorInterface); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if result, err := connection.RequestName(managerName, dbus.NameFlagDoNotQueue); err != nil || result != dbus.RequestNameReplyPrimaryOwner {
|
|
t.Fatalf("manager name: %v %v", result, err)
|
|
}
|
|
return monitor
|
|
}
|
|
|
|
func start_keyboard(t *testing.T, monitor *FakeKeyboardMonitor) *Keyboard {
|
|
t.Helper()
|
|
keyboard := New([]string{"Insert", "KP_Insert"}, input.DesktopKeyboardLayout, nil)
|
|
if err := keyboard.Start(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer cancel()
|
|
if err := keyboard.Stop(ctx); err != nil {
|
|
t.Error(err)
|
|
}
|
|
})
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool { return len(state.modifiers) == 3 })
|
|
return keyboard
|
|
}
|
|
|
|
func wait_snapshot(t *testing.T, monitor *FakeKeyboardMonitor, check func(MonitorSnapshot) bool) MonitorSnapshot {
|
|
t.Helper()
|
|
for {
|
|
select {
|
|
case state := <-monitor.changes:
|
|
if check(state) {
|
|
return state
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("keyboard protocol update timed out")
|
|
}
|
|
}
|
|
}
|
|
|
|
func emit_key(t *testing.T, connection *dbus.Conn, client string, symbol uint32, code uint16, state uint32, released bool) {
|
|
t.Helper()
|
|
message := &dbus.Message{Type: dbus.TypeSignal, Headers: map[dbus.HeaderField]dbus.Variant{
|
|
dbus.FieldDestination: dbus.MakeVariant(client), dbus.FieldPath: dbus.MakeVariant(managerPath),
|
|
dbus.FieldInterface: dbus.MakeVariant(monitorInterface), dbus.FieldMember: dbus.MakeVariant("KeyEvent"),
|
|
dbus.FieldSignature: dbus.MakeVariant(dbus.SignatureOf(released, state, symbol, uint32(0), code)),
|
|
}, Body: []any{released, state, symbol, uint32(0), code}}
|
|
if call := connection.Send(message, make(chan *dbus.Call, 1)); call.Err != nil {
|
|
t.Fatal(call.Err)
|
|
}
|
|
}
|
|
|
|
func (monitor *FakeKeyboardMonitor) key(t *testing.T, symbol uint32, code uint16, state uint32, released bool) {
|
|
monitor.mutex.Lock()
|
|
client := monitor.state.client
|
|
monitor.mutex.Unlock()
|
|
emit_key(t, monitor.connection, client, symbol, code, state, released)
|
|
}
|
|
|
|
func await_command(t *testing.T, keyboard *Keyboard, command input.Command) input.CommandEvent {
|
|
t.Helper()
|
|
select {
|
|
case event := <-keyboard.commands:
|
|
if event.Command != command {
|
|
t.Fatalf("command = %q, want %q", event.Command, command)
|
|
}
|
|
return event
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatalf("command %q timed out", command)
|
|
}
|
|
return input.CommandEvent{}
|
|
}
|
|
|
|
func await_input(t *testing.T, keyboard *Keyboard, name string) input.Event {
|
|
t.Helper()
|
|
select {
|
|
case event := <-keyboard.events:
|
|
if event.Name != name {
|
|
t.Fatalf("input = %q, want %q", event.Name, name)
|
|
}
|
|
return event
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatalf("input %q timed out", name)
|
|
}
|
|
return input.Event{}
|
|
}
|
|
|
|
func TestNativeKeyboardProtocolCommandsMonitoringAndCleanup(t *testing.T) {
|
|
monitor := start_monitor(t)
|
|
keyboard := start_keyboard(t, monitor)
|
|
monitor.key(t, 'x', 53, 0, false)
|
|
monitor.key(t, 'x', 53, 0, true)
|
|
press := await_input(t, keyboard, "x")
|
|
release := await_input(t, keyboard, "x")
|
|
if !press.Pressed || release.Pressed || release.Timestamp <= press.Timestamp {
|
|
t.Fatal("input transition order lost")
|
|
}
|
|
for _, modifierFirst := range []bool{false, true} {
|
|
monitor.key(t, 0xff63, 118, 0, false)
|
|
monitor.key(t, 's', 39, 0, false)
|
|
monitor.key(t, 's', 39, 0, false)
|
|
await_command(t, keyboard, input.CycleSpeechMode)
|
|
if modifierFirst {
|
|
monitor.key(t, 0xff63, 118, 0, true)
|
|
}
|
|
monitor.key(t, 's', 39, 0, true)
|
|
if !modifierFirst {
|
|
monitor.key(t, 0xff63, 118, 0, true)
|
|
}
|
|
monitor.key(t, 'x', 53, 0, false)
|
|
monitor.key(t, 'x', 53, 0, true)
|
|
await_input(t, keyboard, "x")
|
|
await_input(t, keyboard, "x")
|
|
if len(keyboard.commands) != 0 || keyboard.CommandGrabActive() {
|
|
t.Fatal("repeat command or sticky reader modifier")
|
|
}
|
|
}
|
|
monitor.mutex.Lock()
|
|
client := monitor.state.client
|
|
monitor.mutex.Unlock()
|
|
if err := keyboard.Stop(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var exists bool
|
|
if err := monitor.connection.BusObject().Call("org.freedesktop.DBus.NameHasOwner", 0, client).Store(&exists); err != nil || exists {
|
|
t.Fatalf("keyboard connection retained after Stop: %v %v", exists, err)
|
|
}
|
|
}
|
|
|
|
func TestNativeKeyboardBrowseSecureAndLearnTransitions(t *testing.T) {
|
|
monitor := start_monitor(t)
|
|
keyboard := start_keyboard(t, monitor)
|
|
keyboard.SetBrowseNavigation(true)
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool {
|
|
for _, grab := range state.grabs {
|
|
if grab.Symbol == 'h' {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
})
|
|
monitor.key(t, 'h', 43, 0, false)
|
|
monitor.key(t, 'h', 43, 0, true)
|
|
await_command(t, keyboard, input.DocumentNextHeading)
|
|
keyboard.SetNativeButtonFocus(true)
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool {
|
|
for _, grab := range state.grabs {
|
|
if grab.Symbol == 0xff0d {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
})
|
|
keyboard.SetSecureInputActive(true)
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool {
|
|
return !state.watching && len(state.modifiers) == 0 && len(state.grabs) == 0
|
|
})
|
|
keyboard.SetSecureInputActive(false)
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool { return state.watching && len(state.modifiers) == 3 })
|
|
monitor.key(t, 0xff63, 118, 0, false)
|
|
monitor.key(t, 'h', 43, 0, false)
|
|
if !await_command(t, keyboard, input.ToggleLearnMode).LearnModeStarted {
|
|
t.Fatal("learn did not begin")
|
|
}
|
|
monitor.key(t, 'h', 43, 0, true)
|
|
monitor.key(t, 0xff63, 118, 0, true)
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool { return state.grabbed })
|
|
monitor.key(t, 'q', 24, 0, false)
|
|
monitor.key(t, 'q', 24, 0, true)
|
|
await_command(t, keyboard, input.LearnKey)
|
|
monitor.key(t, 0xff1b, 9, 0, false)
|
|
monitor.key(t, 0xff1b, 9, 0, true)
|
|
if await_command(t, keyboard, input.ToggleLearnMode).LearnModeStarted {
|
|
t.Fatal("Escape entered learn mode")
|
|
}
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool { return !state.grabbed })
|
|
keyboard.SetCommandGrabs(false)
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool {
|
|
return len(state.modifiers) == 0 && len(state.grabs) == 0 && !state.grabbed
|
|
})
|
|
monitor.key(t, 'q', 24, 0, false)
|
|
monitor.key(t, 'q', 24, 0, true)
|
|
await_input(t, keyboard, "q")
|
|
await_input(t, keyboard, "q")
|
|
}
|
|
|
|
func TestNativeKeyboardBindingUpdateRollbackAndHeldCancellation(t *testing.T) {
|
|
monitor := start_monitor(t)
|
|
keyboard := start_keyboard(t, monitor)
|
|
monitor.key(t, 0xff63, 118, 0, false)
|
|
monitor.key(t, 's', 39, 0, false)
|
|
await_command(t, keyboard, input.CycleSpeechMode)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
|
defer cancel()
|
|
if err := keyboard.SetKeyboardConfiguration(ctx, input.LaptopKeyboardLayout, []string{"Caps_Lock"}, nil); !errors.Is(err, context.DeadlineExceeded) {
|
|
t.Fatalf("held update = %v", err)
|
|
}
|
|
monitor.key(t, 's', 39, 0, true)
|
|
monitor.key(t, 0xff63, 118, 0, true)
|
|
monitor.mutex.Lock()
|
|
monitor.rejectNext = true
|
|
monitor.mutex.Unlock()
|
|
if err := keyboard.SetKeyboardConfiguration(context.Background(), input.LaptopKeyboardLayout, []string{"Caps_Lock"}, nil); err == nil {
|
|
t.Fatal("rejected update accepted")
|
|
}
|
|
monitor.key(t, 0xff63, 118, 0, false)
|
|
monitor.key(t, ' ', 65, 0, false)
|
|
await_command(t, keyboard, input.Preferences)
|
|
monitor.key(t, ' ', 65, 0, true)
|
|
monitor.key(t, 0xff63, 118, 0, true)
|
|
bindings := []input.BindingOverride{{Layout: input.LaptopKeyboardLayout, Scope: "global", Command: input.Preferences, Keys: []string{"Cthulhu+p"}}}
|
|
if err := keyboard.SetKeyboardConfiguration(context.Background(), input.LaptopKeyboardLayout, []string{"Caps_Lock"}, bindings); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
monitor.key(t, 0xffe5, 66, 0, false)
|
|
monitor.key(t, 'p', 33, 0, false)
|
|
await_command(t, keyboard, input.Preferences)
|
|
monitor.key(t, 'p', 33, 0, true)
|
|
monitor.key(t, 0xffe5, 66, 0, true)
|
|
}
|
|
|
|
func TestNativeKeyboardRejectsSpoofedSignalsAndServiceLoss(t *testing.T) {
|
|
monitor := start_monitor(t)
|
|
keyboard := start_keyboard(t, monitor)
|
|
impostor, err := dbus.ConnectSessionBus()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer impostor.Close()
|
|
monitor.mutex.Lock()
|
|
client := monitor.state.client
|
|
monitor.mutex.Unlock()
|
|
emit_key(t, impostor, client, 0xff63, 118, 0, false)
|
|
if err := impostor.BusObject().Call("org.freedesktop.DBus.GetId", 0).Err; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
monitor.key(t, 's', 39, 0, false)
|
|
monitor.key(t, 's', 39, 0, true)
|
|
await_input(t, keyboard, "s")
|
|
await_input(t, keyboard, "s")
|
|
if len(keyboard.commands) != 0 {
|
|
t.Fatal("untrusted signal created reader modifier")
|
|
}
|
|
monitor.connection.ReleaseName(managerName)
|
|
select {
|
|
case <-keyboard.failures:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("service loss not detected")
|
|
}
|
|
select {
|
|
case <-keyboard.done:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("service loss did not close keyboard connection")
|
|
}
|
|
}
|
|
|
|
func TestNativeKeyboardDeniedWatchCleansUp(t *testing.T) {
|
|
monitor := start_monitor(t)
|
|
monitor.mutex.Lock()
|
|
monitor.rejectWatch = true
|
|
monitor.mutex.Unlock()
|
|
keyboard := New([]string{"Insert"}, input.DesktopKeyboardLayout, nil)
|
|
if err := keyboard.Start(context.Background()); err == nil {
|
|
t.Fatal("denied watch accepted")
|
|
}
|
|
if err := keyboard.Stop(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var exists bool
|
|
if err := monitor.connection.BusObject().Call("org.freedesktop.DBus.NameHasOwner", 0, "org.stormux.Cthulhu.KeyboardMonitor").Store(&exists); err != nil || exists {
|
|
t.Fatal("partial startup retained keyboard identity")
|
|
}
|
|
}
|
|
|
|
func TestNativeKeyboardRapidSecureCycleRetiresHeldModifier(t *testing.T) {
|
|
monitor := start_monitor(t)
|
|
keyboard := start_keyboard(t, monitor)
|
|
monitor.key(t, 0xff63, 118, 0, false)
|
|
monitor.key(t, 't', 28, 0, false)
|
|
await_command(t, keyboard, input.PresentTime)
|
|
keyboard.SetSecureInputActive(true)
|
|
keyboard.SetSecureInputActive(false)
|
|
// The old physical releases may have been discarded while secure. Plain Q
|
|
// must reach monitoring rather than becoming a stale Cthulhu+Q command.
|
|
monitor.key(t, 'q', 24, 0, false)
|
|
monitor.key(t, 'q', 24, 0, true)
|
|
await_input(t, keyboard, "q")
|
|
await_input(t, keyboard, "q")
|
|
if len(keyboard.commands) != 0 || keyboard.CommandGrabActive() {
|
|
t.Fatal("secure cycle retained reader chord")
|
|
}
|
|
}
|
|
|
|
func TestNativeKeyboardStartsSecureWithoutMonitoringOrGrabs(t *testing.T) {
|
|
monitor := start_monitor(t)
|
|
keyboard := New([]string{"Insert"}, input.DesktopKeyboardLayout, nil)
|
|
keyboard.SetSecureInputActive(true)
|
|
if err := keyboard.Start(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer keyboard.Stop(context.Background())
|
|
state := wait_snapshot(t, monitor, func(MonitorSnapshot) bool { return true })
|
|
if state.watching || state.grabbed || len(state.modifiers) != 0 || len(state.grabs) != 0 {
|
|
t.Fatal("secure startup acquired keyboard access")
|
|
}
|
|
keyboard.SetSecureInputActive(false)
|
|
wait_snapshot(t, monitor, func(state MonitorSnapshot) bool { return state.watching && len(state.modifiers) == 1 })
|
|
}
|