289 lines
8.6 KiB
Go
289 lines
8.6 KiB
Go
// Copyright (c) 2026 Storm Dragon
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
package speech
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"net"
|
|
"os"
|
|
"os/exec"
|
|
"os/user"
|
|
"path/filepath"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.stormux.org/storm/fenrir/internal/audio"
|
|
)
|
|
|
|
type disconnectedTransport struct {
|
|
*recordingTransport
|
|
reopen func(context.Context) (Transport, error)
|
|
closed atomic.Bool
|
|
}
|
|
|
|
func (f *disconnectedTransport) Say(string) error { return io.EOF }
|
|
func (f *disconnectedTransport) Cancel() error { return io.EOF }
|
|
func (f *disconnectedTransport) Close() { f.closed.Store(true) }
|
|
func (f *disconnectedTransport) Reopen(ctx context.Context) (Transport, error) {
|
|
return f.reopen(ctx)
|
|
}
|
|
|
|
func TestRecoveryKeepsAdmissionResponsiveAndDropsOutageSpeech(t *testing.T) {
|
|
entered, release := make(chan struct{}), make(chan struct{})
|
|
fresh := new_recording_transport()
|
|
broken := &disconnectedTransport{recordingTransport: new_recording_transport(), reopen: func(ctx context.Context) (Transport, error) {
|
|
close(entered)
|
|
select {
|
|
case <-release:
|
|
return fresh, nil
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
}}
|
|
service := New(broken)
|
|
defer service.Close()
|
|
service.Configure(Parameters{Rate: 40, Pitch: 20, Volume: 80})
|
|
service.Say("lost")
|
|
<-entered
|
|
admitted := make(chan struct{})
|
|
go func() {
|
|
service.Interrupt()
|
|
service.Configure(Parameters{Rate: 70, Pitch: 10, Volume: 90})
|
|
service.Say("discarded during outage")
|
|
service.Toggle_mute()
|
|
service.Say("discarded while muted")
|
|
close(admitted)
|
|
}()
|
|
select {
|
|
case <-admitted:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("audio recovery blocked input admission")
|
|
}
|
|
close(release)
|
|
wait_event(t, fresh.changed, "configure:70:10:90")
|
|
service.Toggle_mute()
|
|
service.Say_with_pitch("fresh", 60)
|
|
wait_event(t, fresh.changed, "sent:fresh")
|
|
fresh.mu.Lock()
|
|
joined := strings.Join(fresh.events, "|")
|
|
fresh.mu.Unlock()
|
|
if strings.Contains(joined, "lost") || strings.Contains(joined, "discarded") || !strings.Contains(joined, "pitch:60|say:fresh|sent:fresh|pitch:10") || !broken.closed.Load() {
|
|
t.Fatal(joined, broken.closed.Load())
|
|
}
|
|
select {
|
|
case err := <-service.Errors():
|
|
t.Fatal("successful recovery was fatal", err)
|
|
default:
|
|
}
|
|
}
|
|
|
|
func TestRecoveryFailureIsFatalAndShutdownCancelsReopen(t *testing.T) {
|
|
for _, failure := range []bool{true, false} {
|
|
t.Run(strconv.FormatBool(failure), func(t *testing.T) {
|
|
entered := make(chan struct{})
|
|
broken := &disconnectedTransport{recordingTransport: new_recording_transport(), reopen: func(ctx context.Context) (Transport, error) {
|
|
close(entered)
|
|
if failure {
|
|
return nil, syscall.EACCES
|
|
}
|
|
<-ctx.Done()
|
|
return nil, ctx.Err()
|
|
}}
|
|
service := New(broken)
|
|
service.Say("trigger recovery")
|
|
<-entered
|
|
if failure {
|
|
select {
|
|
case err := <-service.Errors():
|
|
if !errors.Is(err, syscall.EACCES) {
|
|
t.Fatal(err)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("permanent recovery failure was hidden")
|
|
}
|
|
}
|
|
closed := make(chan struct{})
|
|
go func() { service.Close(); close(closed) }()
|
|
select {
|
|
case <-closed:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("shutdown waited for unavailable speech")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSpeechRecoveryDeadlineReportsFailure(t *testing.T) {
|
|
broken := &disconnectedTransport{recordingTransport: new_recording_transport(), reopen: func(ctx context.Context) (Transport, error) {
|
|
<-ctx.Done()
|
|
return nil, ctx.Err()
|
|
}}
|
|
service := New(broken)
|
|
defer service.Close()
|
|
ctx, cancel := context.WithTimeout(service.recoveryContext, 40*time.Millisecond)
|
|
defer cancel()
|
|
service.recoveryContext = ctx
|
|
service.Say("trigger unavailable daemon")
|
|
select {
|
|
case err := <-service.Errors():
|
|
if !errors.Is(err, context.DeadlineExceeded) {
|
|
t.Fatal(err)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("exhausted recovery left the reader silently running")
|
|
}
|
|
}
|
|
|
|
func TestOpenContextCancelsSilentHandshakeAndDoesNotOwnLiveClient(t *testing.T) {
|
|
listener, err := net.Listen("unix", filepath.Join(t.TempDir(), "silent.sock"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer listener.Close()
|
|
accepted := make(chan net.Conn, 1)
|
|
go func() { conn, _ := listener.Accept(); accepted <- conn }()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
|
|
defer cancel()
|
|
started := time.Now()
|
|
client, err := Open_driver_context(ctx, "speech-dispatcher", Options{Socket: listener.Addr().String()})
|
|
if client != nil || err == nil || time.Since(started) > time.Second {
|
|
t.Fatal("silent handshake ignored cancellation", client, err)
|
|
}
|
|
(<-accepted).Close()
|
|
server := new_fake_server(t, false)
|
|
ctx, cancel = context.WithCancel(context.Background())
|
|
client, err = Open_driver_context(ctx, "speech-dispatcher", Options{Socket: server.path})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer client.Close()
|
|
cancel()
|
|
if err := client.Say("live client survives startup context"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestSudoSpawnDropsChildCredentialsAndFiltersRootEnvironment(t *testing.T) {
|
|
t.Setenv("HOME", "/root")
|
|
t.Setenv("XDG_RUNTIME_DIR", "/root/runtime")
|
|
t.Setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/root/session-bus")
|
|
t.Setenv("FENRIR_TEST_SECRET", "must-not-reach-child")
|
|
t.Setenv("PULSE_SERVER", "unix:/tmp/selected-audio.sock")
|
|
command := exec.Command("speech-dispatcher", "--spawn")
|
|
account := &user.User{Uid: "2000", Gid: "2001", Username: "Username", HomeDir: "/home/Username"}
|
|
if err := configure_sudo_spawn(command, account, []string{"2001", "2002"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want := &syscall.Credential{Uid: 2000, Gid: 2001, Groups: []uint32{2001, 2002}}
|
|
if !reflect.DeepEqual(command.SysProcAttr.Credential, want) || command.Dir != account.HomeDir {
|
|
t.Fatal(command.SysProcAttr, command.Dir)
|
|
}
|
|
environment := strings.Join(command.Env, "\n")
|
|
if strings.Contains(environment, "/root") || strings.Contains(environment, "FENRIR_TEST_SECRET") || !strings.Contains(environment, "HOME=/home/Username") || !strings.Contains(environment, "PULSE_SERVER=unix:/tmp/selected-audio.sock") {
|
|
t.Fatal(environment)
|
|
}
|
|
account.Uid = "0"
|
|
if err := configure_sudo_spawn(command, account, nil); err == nil {
|
|
t.Fatal("invalid invoking user selected root")
|
|
}
|
|
}
|
|
|
|
func TestStartupWithoutSudoRetriesUntilPrivateDaemonAppears(t *testing.T) {
|
|
t.Setenv("SUDO_UID", "")
|
|
t.Setenv("SPEECHD_CMD", "/nonexistent-fixture-speech-dispatcher")
|
|
socket := filepath.Join(t.TempDir(), "speech.sock")
|
|
t.Setenv("SPEECHD_ADDRESS", "unix_socket:"+socket)
|
|
t.Setenv("SPEECHD_SOCKET", "")
|
|
uid := os.Geteuid()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
failed := make(chan struct{}, 1)
|
|
result := make(chan error, 1)
|
|
var client Transport
|
|
go func() {
|
|
result <- audio.Retry(ctx, func(ctx context.Context) error {
|
|
var err error
|
|
client, err = Open_driver_context(ctx, "speech-dispatcher", Options{})
|
|
if err != nil {
|
|
select {
|
|
case failed <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
return err
|
|
}, audio.Temporary)
|
|
}()
|
|
select {
|
|
case <-failed:
|
|
case <-ctx.Done():
|
|
t.Fatal("initial absent daemon was not detected")
|
|
}
|
|
listener, err := net.Listen("unix", socket)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
server := new_fake_server_listener(t, false, listener)
|
|
if err := <-result; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer client.Close()
|
|
cancel()
|
|
if os.Geteuid() != uid {
|
|
t.Fatal("startup changed reader credentials")
|
|
}
|
|
if err := client.Say("late daemon ready"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
wait_event(t, server.events, "say:late daemon ready")
|
|
}
|
|
|
|
func TestClientReconnectsAfterPrivateServerDropsBothConnections(t *testing.T) {
|
|
server := new_fake_server(t, false)
|
|
client, err := Open(Options{Socket: server.path})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
service := New(client)
|
|
defer service.Close()
|
|
service.Configure(Parameters{Rate: 30, Pitch: 10, Volume: 70})
|
|
service.Say("before disconnect")
|
|
wait_event(t, server.events, "say:before disconnect")
|
|
server.mu.Lock()
|
|
for _, conn := range server.connections {
|
|
conn.Close()
|
|
}
|
|
server.mu.Unlock()
|
|
service.Say("failed request")
|
|
deadline := time.NewTimer(time.Second)
|
|
defer deadline.Stop()
|
|
tick := time.NewTicker(time.Millisecond)
|
|
defer tick.Stop()
|
|
for {
|
|
service.mu.Lock()
|
|
service.transportMu.RLock()
|
|
restored := service.transport != nil && service.transport != client && !service.recovering
|
|
service.transportMu.RUnlock()
|
|
service.mu.Unlock()
|
|
if restored {
|
|
break
|
|
}
|
|
select {
|
|
case err := <-service.Errors():
|
|
t.Fatal("reconnection failed", err)
|
|
case <-deadline.C:
|
|
t.Fatal("both Speech Dispatcher sockets were not restored")
|
|
case <-tick.C:
|
|
}
|
|
}
|
|
service.Say("after reconnect")
|
|
wait_event(t, server.events, "say:after reconnect")
|
|
}
|