314 lines
8.4 KiB
Go
314 lines
8.4 KiB
Go
// Copyright (c) 2026 Storm Dragon
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
package tentacle
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"os"
|
|
"strconv"
|
|
"sync"
|
|
"syscall"
|
|
"time"
|
|
|
|
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
|
|
)
|
|
|
|
type SessionEvent = tentaclewire.SessionEvent
|
|
type RemoteOpen = tentaclewire.RemoteOpen
|
|
type RemoteLease = tentaclewire.RemoteLease
|
|
type RemoteOperation = tentaclewire.RemoteOperation
|
|
type RemoteSpeechPart = tentaclewire.RemoteSpeechPart
|
|
type RemoteSpeech = tentaclewire.RemoteSpeech
|
|
type RemotePresentation = tentaclewire.RemotePresentation
|
|
|
|
const MaxRemoteSpeechBytes = tentaclewire.MaxRemoteSpeechBytes
|
|
|
|
func ValidRemoteSpeech(speech RemoteSpeech) bool { return tentaclewire.ValidRemoteSpeech(speech) }
|
|
|
|
func ValidRemotePresentation(request RemotePresentation) bool {
|
|
return tentaclewire.ValidRemotePresentation(request)
|
|
}
|
|
|
|
type SessionError struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
}
|
|
|
|
func (err *SessionError) Error() string { return "tentacle session error " + strconv.Itoa(err.Code) }
|
|
|
|
type sessionReply struct {
|
|
value json.RawMessage
|
|
err error
|
|
}
|
|
|
|
// SessionClient is single-use. ServeSession owns its bounded writer and receive
|
|
// loop; command handlers and background workers may Call concurrently. Events
|
|
// must be drained without blocking on network I/O. Overflow retires the process
|
|
// session rather than dropping lifecycle events. No desktop access is implicit.
|
|
type SessionClient struct {
|
|
mu sync.Mutex
|
|
ctx context.Context
|
|
abort func()
|
|
active bool
|
|
started bool
|
|
sequence uint64
|
|
eventSequence uint64
|
|
pending map[string]chan sessionReply
|
|
events chan SessionEvent
|
|
outbound chan []byte
|
|
queuedBytes int
|
|
}
|
|
|
|
func NewSessionClient() *SessionClient {
|
|
return &SessionClient{events: make(chan SessionEvent, 32), outbound: make(chan []byte, 32), pending: make(map[string]chan sessionReply)}
|
|
}
|
|
|
|
func (client *SessionClient) Events() <-chan SessionEvent { return client.events }
|
|
|
|
// ServeSession opts into protocol 1.1; a protocol 1.0 host is rejected.
|
|
func ServeSession(input io.ReadCloser, output io.WriteCloser, commands []Command, extensions *Extensions, client *SessionClient) error {
|
|
if client == nil {
|
|
return ErrProtocol
|
|
}
|
|
if extensions != nil {
|
|
if err := tentaclewire.ValidateExtensions(*extensions); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// os.Stdin may be created before the runtime marks a pipe pollable. A
|
|
// blocking read on it cannot reliably be interrupted by Close from another
|
|
// goroutine. Adopt owned nonblocking duplicates so Go's poller can wake both
|
|
// pipe directions on timeout. Non-file closers (e.g. io.Pipe) retain their
|
|
// documented Close behavior.
|
|
if file, ok := input.(*os.File); ok {
|
|
pollable, err := sessionPipe(file)
|
|
if err != nil {
|
|
input.Close()
|
|
output.Close()
|
|
return err
|
|
}
|
|
input = pollable
|
|
}
|
|
if file, ok := output.(*os.File); ok {
|
|
pollable, err := sessionPipe(file)
|
|
if err != nil {
|
|
input.Close()
|
|
output.Close()
|
|
return err
|
|
}
|
|
output = pollable
|
|
}
|
|
return serve(input, output, commands, extensions, client)
|
|
}
|
|
|
|
func sessionPipe(file *os.File) (*os.File, error) {
|
|
fd, err := syscall.Dup(int(file.Fd()))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
syscall.CloseOnExec(fd)
|
|
if err := syscall.SetNonblock(fd, true); err != nil {
|
|
syscall.Close(fd)
|
|
return nil, err
|
|
}
|
|
owned := os.NewFile(uintptr(fd), file.Name())
|
|
file.Close()
|
|
return owned, nil
|
|
}
|
|
|
|
func (client *SessionClient) start(ctx context.Context, abort func()) bool {
|
|
client.mu.Lock()
|
|
defer client.mu.Unlock()
|
|
if client.started || client.events == nil {
|
|
return false
|
|
}
|
|
client.started, client.ctx, client.abort = true, ctx, abort
|
|
return true
|
|
}
|
|
|
|
func (client *SessionClient) enqueue(frame []byte) error {
|
|
client.mu.Lock()
|
|
defer client.mu.Unlock()
|
|
return client.enqueueLocked(frame)
|
|
}
|
|
|
|
func (client *SessionClient) enqueueLocked(frame []byte) error {
|
|
if client.ctx == nil || client.ctx.Err() != nil {
|
|
return io.ErrClosedPipe
|
|
}
|
|
if client.queuedBytes+len(frame) > tentaclewire.SessionQueueBytes {
|
|
client.abort()
|
|
return tentaclewire.ErrLimit
|
|
}
|
|
select {
|
|
case client.outbound <- frame:
|
|
client.queuedBytes += len(frame)
|
|
return nil
|
|
default:
|
|
client.abort()
|
|
return tentaclewire.ErrLimit
|
|
}
|
|
}
|
|
|
|
func (client *SessionClient) write(output io.Writer) {
|
|
for {
|
|
select {
|
|
case <-client.ctx.Done():
|
|
return
|
|
case frame := <-client.outbound:
|
|
if client.ctx.Err() != nil {
|
|
return
|
|
}
|
|
n, err := output.Write(frame)
|
|
if err == nil && n != len(frame) {
|
|
err = io.ErrShortWrite
|
|
}
|
|
client.mu.Lock()
|
|
client.queuedBytes -= len(frame)
|
|
client.mu.Unlock()
|
|
if err != nil {
|
|
client.abort()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Call has a two-second maximum deadline. Timeout/cancellation retires the
|
|
// session, so an abandoned effect cannot leave unbounded pending calls behind.
|
|
// Typed desktop operations require separately attached reader owners; a
|
|
// successful call acknowledges admission, not delivery to a desktop/backend.
|
|
func (client *SessionClient) Call(ctx context.Context, method string, params any, result any) error {
|
|
if ctx == nil {
|
|
return errors.New("nil session call context")
|
|
}
|
|
ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
|
defer cancel()
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
data, err := json.Marshal(params)
|
|
if err != nil || len(data) > 8192 || tentaclewire.ValidateJSON(data) != nil || len(method) == 0 || len(method) > 128 {
|
|
return ErrProtocol
|
|
}
|
|
client.mu.Lock()
|
|
if !client.active || client.ctx.Err() != nil {
|
|
client.mu.Unlock()
|
|
return io.ErrClosedPipe
|
|
}
|
|
if len(client.pending) >= 8 {
|
|
client.mu.Unlock()
|
|
return tentaclewire.ErrLimit
|
|
}
|
|
client.sequence++
|
|
id := "c:" + strconv.FormatUint(client.sequence, 10)
|
|
frame, err := json.Marshal(request{JSONRPC: "2.0", ID: id, Method: method, Params: data})
|
|
if err != nil {
|
|
client.mu.Unlock()
|
|
return ErrProtocol
|
|
}
|
|
response := make(chan sessionReply, 1)
|
|
client.pending[id] = response
|
|
err = client.enqueueLocked(append(frame, '\n'))
|
|
client.mu.Unlock()
|
|
defer func() { client.mu.Lock(); delete(client.pending, id); client.mu.Unlock() }()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
client.abort()
|
|
return ctx.Err()
|
|
case <-client.ctx.Done():
|
|
return io.ErrClosedPipe
|
|
case reply := <-response:
|
|
if ctx.Err() != nil {
|
|
client.abort()
|
|
return ctx.Err()
|
|
}
|
|
if client.ctx.Err() != nil {
|
|
return io.ErrClosedPipe
|
|
}
|
|
if reply.err != nil {
|
|
return reply.err
|
|
}
|
|
if result != nil {
|
|
return tentaclewire.DecodeStrict(reply.value, result)
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (client *SessionClient) acceptReply(frame []byte) error {
|
|
var reply struct {
|
|
JSONRPC string `json:"jsonrpc"`
|
|
ID string `json:"id"`
|
|
Result json.RawMessage `json:"result"`
|
|
Error *SessionError `json:"error"`
|
|
}
|
|
if tentaclewire.DecodeStrict(frame, &reply) != nil || reply.JSONRPC != "2.0" {
|
|
return ErrProtocol
|
|
}
|
|
var fields map[string]json.RawMessage
|
|
_ = json.Unmarshal(frame, &fields)
|
|
_, hasResult := fields["result"]
|
|
_, hasError := fields["error"]
|
|
if hasResult == hasError || hasError && (reply.Error == nil || len(reply.Error.Message) > 1024) {
|
|
return ErrProtocol
|
|
}
|
|
if hasError {
|
|
var fields map[string]json.RawMessage
|
|
_ = json.Unmarshal(frame, &fields)
|
|
var detail map[string]json.RawMessage
|
|
_ = json.Unmarshal(fields["error"], &detail)
|
|
if len(detail["code"]) == 0 || string(detail["code"]) == "null" || len(detail["message"]) == 0 || string(detail["message"]) == "null" {
|
|
return ErrProtocol
|
|
}
|
|
}
|
|
client.mu.Lock()
|
|
defer client.mu.Unlock()
|
|
waiter := client.pending[reply.ID]
|
|
if !client.active || waiter == nil {
|
|
return ErrProtocol
|
|
}
|
|
delete(client.pending, reply.ID)
|
|
var err error
|
|
if reply.Error != nil {
|
|
err = reply.Error
|
|
}
|
|
waiter <- sessionReply{value: reply.Result, err: err}
|
|
return nil
|
|
}
|
|
|
|
func (client *SessionClient) acceptEvent(params json.RawMessage) error {
|
|
var event SessionEvent
|
|
var fields map[string]json.RawMessage
|
|
if json.Unmarshal(params, &fields) != nil {
|
|
return ErrProtocol
|
|
}
|
|
for _, value := range fields {
|
|
if string(value) == "null" {
|
|
return ErrProtocol
|
|
}
|
|
}
|
|
if tentaclewire.DecodeStrict(params, &event) != nil || !tentaclewire.ValidSessionEvent(event) {
|
|
return ErrProtocol
|
|
}
|
|
client.mu.Lock()
|
|
defer client.mu.Unlock()
|
|
if !client.active || event.Sequence != client.eventSequence+1 {
|
|
return ErrProtocol
|
|
}
|
|
client.eventSequence = event.Sequence
|
|
select {
|
|
case client.events <- event:
|
|
return nil
|
|
default:
|
|
return tentaclewire.ErrLimit
|
|
}
|
|
}
|