337 lines
12 KiB
Go
337 lines
12 KiB
Go
// Copyright (c) 2026 Storm Dragon
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
package tentacle
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.stormux.org/storm/cthulhu/internal/tentaclewire"
|
|
)
|
|
|
|
const RemoteHeartbeatTimeout = 3 * time.Second
|
|
const remoteApprovalTimeout = 30 * time.Second
|
|
|
|
var ErrPermission = errors.New("tentacle permission denied")
|
|
|
|
// RemoteGrants returns install permissions, never current consent.
|
|
func (session *Session) RemoteGrants() []string { return slices.Clone(session.spec.RemoteGrants) }
|
|
|
|
// Owned by Session.mutex. No network identity or secret belongs here.
|
|
type remoteAccess struct {
|
|
action string
|
|
lease string
|
|
epoch uint64
|
|
sequence uint64
|
|
role string
|
|
grants []string
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
deadline time.Time
|
|
timer *time.Timer
|
|
stopPolicy func() bool
|
|
speechBatch *remoteSpeechBatch
|
|
speechItems int
|
|
speechBytes int
|
|
keyItems int
|
|
keyBytes int
|
|
}
|
|
|
|
// AuthorizeRemote is the unrestricted protocol-fixture hook. Reader managers
|
|
// use RemotePolicy and AuthorizeRemoteWithApproval instead; child RPCs cannot
|
|
// call either authorizer.
|
|
func (session *Session) AuthorizeRemote(role string, grants []string) (tentaclewire.SessionEvent, error) {
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
if session.manager.remotePolicy != nil {
|
|
return tentaclewire.SessionEvent{}, ErrPermission
|
|
}
|
|
return session.authorize_remote_locked(role, grants, session.outputContext)
|
|
}
|
|
|
|
func (session *Session) authorize_remote_locked(role string, grants []string, parent context.Context) (tentaclewire.SessionEvent, error) {
|
|
if session.state != Active || session.ctx.Err() != nil || !session.spec.SessionEvents {
|
|
return tentaclewire.SessionEvent{}, ErrUnavailable
|
|
}
|
|
if !tentaclewire.ValidRoleGrants(role, grants) {
|
|
return tentaclewire.SessionEvent{}, ErrPermission
|
|
}
|
|
for _, grant := range grants {
|
|
if !slices.Contains(session.spec.RemoteGrants, grant) {
|
|
return tentaclewire.SessionEvent{}, ErrPermission
|
|
}
|
|
}
|
|
var token [16]byte
|
|
if _, err := rand.Read(token[:]); err != nil {
|
|
return tentaclewire.SessionEvent{}, err
|
|
}
|
|
session.manager.remoteMutex.Lock()
|
|
if owner := session.manager.remoteOwner; owner != nil && owner != session {
|
|
session.manager.remoteMutex.Unlock()
|
|
return tentaclewire.SessionEvent{}, ErrBusy
|
|
}
|
|
session.manager.remoteMutex.Unlock()
|
|
session.revoke_remote_locked("replaced", true)
|
|
// Claim and check atomically: a different session may have acquired the slot
|
|
// while the old lease was being revoked.
|
|
session.manager.remoteMutex.Lock()
|
|
if session.manager.remoteOwner != nil {
|
|
session.manager.remoteMutex.Unlock()
|
|
return tentaclewire.SessionEvent{}, ErrBusy
|
|
}
|
|
session.manager.remoteOwner = session
|
|
session.manager.remoteMutex.Unlock()
|
|
if session.state != Active {
|
|
session.release_remote_owner()
|
|
return tentaclewire.SessionEvent{}, ErrUnavailable
|
|
}
|
|
session.remoteEpoch++
|
|
ctx, cancel := context.WithCancel(parent)
|
|
access := &remoteAccess{action: hex.EncodeToString(token[:]), epoch: session.remoteEpoch,
|
|
role: role, grants: slices.Clone(grants), ctx: ctx, cancel: cancel}
|
|
session.remote = access
|
|
// Parent cancellation invalidates delivery immediately. Retirement/event
|
|
// cleanup acquires the session lock afterwards and cannot retire a newer epoch.
|
|
access.stopPolicy = context.AfterFunc(parent, func() {
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
if session.remote == access {
|
|
session.revoke_remote_locked("local", true)
|
|
}
|
|
})
|
|
session.refresh_remote_locked(remoteApprovalTimeout)
|
|
event := tentaclewire.SessionEvent{Kind: "remote.action", Action: access.action, Epoch: access.epoch, Role: role, Grants: slices.Clone(grants)}
|
|
if !session.remote_event_locked(&event) {
|
|
session.fail_locked(ErrLimit)
|
|
return tentaclewire.SessionEvent{}, ErrLimit
|
|
}
|
|
return event, nil
|
|
}
|
|
|
|
// RevokeRemote invalidates buffered work synchronously. A later local consent
|
|
// action is required; heartbeat and remote.open cannot revive the old epoch.
|
|
func (session *Session) RevokeRemote() {
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
session.revoke_remote_locked("local", true)
|
|
}
|
|
|
|
// RevokeRemoteEpoch retires only the action/lease owned by a completed core
|
|
// interaction. Closing an old workflow cannot revoke a replacement approval.
|
|
func (session *Session) RevokeRemoteEpoch(epoch uint64) {
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
if session.remote != nil && session.remote.epoch == epoch {
|
|
session.revoke_remote_locked("local", true)
|
|
}
|
|
}
|
|
|
|
// RemoteContext is for future reader owners to check again at delivery. A live
|
|
// context is necessary, not sufficient: focus, lock and output policy still apply.
|
|
func (session *Session) RemoteContext(lease tentaclewire.RemoteLease, grant string) (context.Context, error) {
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
access := session.remote_access_locked(lease, grant)
|
|
if access == nil {
|
|
return nil, ErrPermission
|
|
}
|
|
return access.ctx, nil
|
|
}
|
|
|
|
// RemoteLeaseContext checks process ownership and lifetime, without authorizing
|
|
// any desktop effect. Core uses it when returning from a connection form to the
|
|
// desktop; effect owners must still use RemoteContext with their specific grant.
|
|
func (session *Session) RemoteLeaseContext(lease tentaclewire.RemoteLease) (context.Context, error) {
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
access := session.remote_lease_locked(lease)
|
|
if access == nil {
|
|
return nil, ErrPermission
|
|
}
|
|
return access.ctx, nil
|
|
}
|
|
|
|
func (session *Session) release_remote_owner() {
|
|
session.manager.remoteMutex.Lock()
|
|
if session.manager.remoteOwner == session {
|
|
session.manager.remoteOwner = nil
|
|
}
|
|
session.manager.remoteMutex.Unlock()
|
|
}
|
|
|
|
func (session *Session) revoke_remote_locked(reason string, notify bool) {
|
|
if session.remoteApproval != nil {
|
|
session.remoteApproval.cancel()
|
|
session.remoteApproval = nil
|
|
}
|
|
access := session.remote
|
|
if access == nil {
|
|
return
|
|
}
|
|
session.remote = nil
|
|
access.cancel()
|
|
access.speechBatch = nil
|
|
access.speechItems, access.speechBytes = 0, 0
|
|
access.keyItems, access.keyBytes = 0, 0
|
|
if session.remoteSpeech != nil {
|
|
drain:
|
|
for {
|
|
select {
|
|
case <-session.remoteSpeech:
|
|
default:
|
|
break drain
|
|
}
|
|
}
|
|
}
|
|
if session.remoteKeys != nil {
|
|
drainKeys:
|
|
for {
|
|
select {
|
|
case <-session.remoteKeys:
|
|
default:
|
|
break drainKeys
|
|
}
|
|
}
|
|
}
|
|
if access.stopPolicy != nil {
|
|
access.stopPolicy()
|
|
}
|
|
if access.timer != nil {
|
|
access.timer.Stop()
|
|
}
|
|
session.release_remote_owner()
|
|
if notify {
|
|
event := tentaclewire.SessionEvent{Kind: "remote.revoked", Action: access.action, Lease: access.lease, Epoch: access.epoch, Reason: reason}
|
|
if !session.remote_event_locked(&event) {
|
|
session.fail_locked(ErrLimit)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (session *Session) refresh_remote_locked(duration time.Duration) {
|
|
access := session.remote
|
|
access.deadline = time.Now().Add(duration)
|
|
if access.timer != nil {
|
|
access.timer.Stop()
|
|
}
|
|
access.timer = time.AfterFunc(duration, func() {
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
// An old timer may already be running when heartbeat resets it.
|
|
if session.remote == access && !time.Now().Before(access.deadline) {
|
|
session.revoke_remote_locked("expired", true)
|
|
}
|
|
})
|
|
}
|
|
|
|
func (session *Session) remote_event_locked(event *tentaclewire.SessionEvent) bool {
|
|
session.eventSequence++
|
|
event.Sequence = session.eventSequence
|
|
return session.notify_locked("session.event", event)
|
|
}
|
|
|
|
func (session *Session) accept_child_request(frame []byte) error {
|
|
var message struct {
|
|
JSONRPC string `json:"jsonrpc"`
|
|
ID string `json:"id"`
|
|
Method string `json:"method"`
|
|
Params json.RawMessage `json:"params"`
|
|
}
|
|
if decode_strict(frame, &message) != nil || message.JSONRPC != "2.0" || !strings.HasPrefix(message.ID, "c:") || message.Method == "" || len(message.Method) > 128 || len(message.Params) == 0 || len(message.Params) > 8192 {
|
|
return ErrProtocol
|
|
}
|
|
number, err := strconv.ParseUint(strings.TrimPrefix(message.ID, "c:"), 10, 64)
|
|
if err != nil || message.ID != "c:"+strconv.FormatUint(number, 10) {
|
|
return ErrProtocol
|
|
}
|
|
session.mutex.Lock()
|
|
defer session.mutex.Unlock()
|
|
if !session.spec.SessionEvents || session.state != Active || session.ctx.Err() != nil || number <= session.childSequence {
|
|
return ErrProtocol
|
|
}
|
|
session.childSequence = number
|
|
value, failure := session.remote_call_locked(message.Method, message.Params)
|
|
reply := map[string]any{"jsonrpc": "2.0", "id": message.ID}
|
|
if failure != nil {
|
|
reply["error"] = failure
|
|
} else {
|
|
reply["result"] = value
|
|
}
|
|
data, err := json.Marshal(reply)
|
|
if err != nil || !session.enqueue_locked(outgoing{frame: append(data, '\n')}) {
|
|
return ErrLimit
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (session *Session) remote_call_locked(method string, params json.RawMessage) (any, *RPCError) {
|
|
denied := &RPCError{Code: -32001, Message: "Session permission denied"}
|
|
invalid := &RPCError{Code: -32602, Message: "Invalid session parameters"}
|
|
switch method {
|
|
case "remote.present":
|
|
return session.remote_present_locked(params)
|
|
case "remote.key":
|
|
return session.remote_key_locked(params)
|
|
case "remote.open":
|
|
var request tentaclewire.RemoteOpen
|
|
if decode_strict(params, &request) != nil || len(request.Action) != 32 || request.Epoch == 0 || !tentaclewire.ValidRoleGrants(request.Role, request.Grants) {
|
|
return nil, invalid
|
|
}
|
|
access := session.remote
|
|
if access == nil || access.lease != "" || access.action != request.Action || access.epoch != request.Epoch || access.role != request.Role || !time.Now().Before(access.deadline) || access.ctx.Err() != nil {
|
|
return nil, denied
|
|
}
|
|
for _, grant := range request.Grants {
|
|
if !slices.Contains(access.grants, grant) {
|
|
return nil, denied
|
|
}
|
|
}
|
|
var token [16]byte
|
|
if _, err := rand.Read(token[:]); err != nil {
|
|
return nil, &RPCError{Code: -32603, Message: "Session unavailable"}
|
|
}
|
|
access.lease, access.action, access.grants = hex.EncodeToString(token[:]), "", slices.Clone(request.Grants)
|
|
session.refresh_remote_locked(RemoteHeartbeatTimeout)
|
|
return tentaclewire.RemoteLease{Lease: access.lease, Epoch: access.epoch}, nil
|
|
case "remote.close", "remote.heartbeat":
|
|
var request tentaclewire.RemoteOperation
|
|
if decode_strict(params, &request) != nil || len(request.Lease) != 32 || request.Epoch == 0 || request.Sequence == 0 {
|
|
return nil, invalid
|
|
}
|
|
access := session.remote
|
|
// Closing an already retired epoch has no effect on a newer lease.
|
|
if method == "remote.close" && request.Epoch <= session.remoteEpoch && (access == nil || access.epoch != request.Epoch) {
|
|
return struct{}{}, nil
|
|
}
|
|
if access == nil || access.lease == "" || access.lease != request.Lease || access.epoch != request.Epoch || request.Sequence <= access.sequence {
|
|
return nil, denied
|
|
}
|
|
// Closing only removes authority. Allow authenticated cleanup even
|
|
// before the expiry/policy callback acquires this lock; rejecting it
|
|
// would make the child permanently distrust an otherwise healthy IPC
|
|
// session. Heartbeats must never renew that cancelled authority.
|
|
if method == "remote.heartbeat" && (!time.Now().Before(access.deadline) || access.ctx.Err() != nil) {
|
|
return nil, denied
|
|
}
|
|
access.sequence = request.Sequence
|
|
if method == "remote.close" {
|
|
session.revoke_remote_locked("closed", true)
|
|
} else {
|
|
session.refresh_remote_locked(RemoteHeartbeatTimeout)
|
|
}
|
|
return struct{}{}, nil
|
|
default:
|
|
// No generic callback or desktop effect path exists. New operations need
|
|
// typed validation and owner integration, never arbitrary method dispatch.
|
|
return nil, &RPCError{Code: -32601, Message: "Unsupported session operation"}
|
|
}
|
|
}
|