Reticulum-Go/pkg/interfaces/pipe.go

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-2026 Quad4.io

package interfaces

import (
	"fmt"
	"io"
	"net"
	"os/exec"
	"strings"
	"sync"
	"time"

	"quad4/reticulum-go/pkg/common"
	"quad4/reticulum-go/pkg/debug"
	"quad4/reticulum-go/pkg/sandbox"
)

const (
	pipeHWMTU         = 1064
	pipeBitrateGuess  = 1_000_000
	defaultRespawnDly = 5 * time.Second
)

// PipeInterface bridges Reticulum to an external program over stdin/stdout
// with HDLC framing.
type PipeInterface struct {
	BaseInterface
	command      string
	respawnDelay time.Duration
	cmd          *exec.Cmd
	stdin        io.WriteCloser
	stdout       io.ReadCloser
	done         chan struct{}
	stopOnce     sync.Once
	respawning   bool
	panicOnError bool
	txMu         sync.Mutex
	txFrame      []byte
	readBuf      []byte
}

// NewPipeInterface starts a subprocess identified by command with HDLC framing.
func NewPipeInterface(name, command string, enabled bool, respawnDelay time.Duration, panicOnError bool) (*PipeInterface, error) {
	if strings.TrimSpace(command) == "" {
		return nil, fmt.Errorf("no command specified for PipeInterface")
	}
	if respawnDelay <= 0 {
		respawnDelay = defaultRespawnDly
	}
	pi := &PipeInterface{
		BaseInterface: NewBaseInterface(name, common.IFTypePipe, enabled),
		command:       command,
		respawnDelay:  respawnDelay,
		done:          make(chan struct{}),
		panicOnError:  panicOnError,
		txFrame:       make([]byte, 0, pipeHWMTU*2+4),
		readBuf:       make([]byte, streamReadChunk),
	}
	pi.In = true
	pi.Out = true
	pi.MTU = pipeHWMTU
	pi.Bitrate = pipeBitrateGuess
	if enabled {
		if err := pi.openPipe(); err != nil {
			return nil, err
		}
		pi.startReadLoop()
	}
	return pi, nil
}

func (pi *PipeInterface) String() string {
	return fmt.Sprintf("PipeInterface[%s]", pi.Name)
}

func (pi *PipeInterface) Start() error {
	pi.Mutex.Lock()
	if pi.Online {
		pi.Mutex.Unlock()
		return nil
	}
	enabled := pi.Enabled
	pi.Mutex.Unlock()
	if !enabled {
		return fmt.Errorf("interface not enabled")
	}
	if err := pi.openPipe(); err != nil {
		return err
	}
	pi.startReadLoop()
	return nil
}

func (pi *PipeInterface) Stop() error {
	pi.Mutex.Lock()
	pi.Enabled = false
	pi.Online = false
	pi.Mutex.Unlock()
	pi.stopOnce.Do(func() {
		close(pi.done)
	})
	pi.killProcess()
	return nil
}

func (pi *PipeInterface) openPipe() error {
	args, err := splitPipeCommand(pi.command)
	if err != nil {
		return err
	}
	cmd := exec.Command(args[0], args[1:]...) // #nosec G204 -- user-configured pipe command
	stdin, err := cmd.StdinPipe()
	if err != nil {
		return err
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		_ = stdin.Close()
		return err
	}
	if err := sandbox.StartLimited(cmd); err != nil {
		_ = stdin.Close()
		_ = stdout.Close()
		return err
	}
	pi.Mutex.Lock()
	pi.killLocked()
	pi.cmd = cmd
	pi.stdin = stdin
	pi.stdout = stdout
	pi.Online = true
	pi.Mutex.Unlock()
	debug.Log(debug.DebugVerbose, "Subprocess pipe connected", "name", pi.Name)
	return nil
}

func (pi *PipeInterface) killProcess() {
	pi.Mutex.Lock()
	defer pi.Mutex.Unlock()
	pi.killLocked()
}

func (pi *PipeInterface) killLocked() {
	if pi.stdin != nil {
		_ = pi.stdin.Close()
		pi.stdin = nil
	}
	if pi.stdout != nil {
		_ = pi.stdout.Close()
		pi.stdout = nil
	}
	if pi.cmd != nil && pi.cmd.Process != nil {
		_ = pi.cmd.Process.Kill()
		_, _ = pi.cmd.Process.Wait()
	}
	pi.cmd = nil
}

func (pi *PipeInterface) ProcessOutgoing(data []byte) error {
	pi.txMu.Lock()
	defer pi.txMu.Unlock()

	pi.Mutex.RLock()
	online := pi.Online
	stdin := pi.stdin
	pi.Mutex.RUnlock()
	if !online || stdin == nil {
		return fmt.Errorf("pipe interface offline")
	}
	frame := appendFrameHDLC(pi.txFrame[:0], data)
	pi.txFrame = frame
	n, err := stdin.Write(frame)
	if err != nil {
		pi.handleIOError(err)
		return err
	}
	if n != len(frame) {
		return fmt.Errorf("pipe interface only wrote %d bytes of %d", n, len(frame))
	}
	return nil
}

func (pi *PipeInterface) Send(data []byte, address string) error {
	if err := common.RejectReceiveOnly(pi); err != nil {
		return err
	}
	masked, err := common.ApplyIFACOutbound(pi, data)
	if err != nil {
		return err
	}
	if err := pi.ProcessOutgoing(masked); err != nil {
		return err
	}
	pi.updateBandwidthStats(uint64(len(masked)))
	return nil
}

func (pi *PipeInterface) startReadLoop() {
	go pi.readLoop()
}

func (pi *PipeInterface) readLoop() {
	decoder := newHDLCStreamDecoder(pi.MTU, pi.ProcessIncoming)
	n := streamReadSize(pi.MTU)
	buffer := pi.readBuf
	if cap(buffer) < n {
		buffer = make([]byte, n)
		pi.readBuf = buffer
	}
	buffer = buffer[:n]

	for {
		pi.Mutex.RLock()
		stdout := pi.stdout
		done := pi.done
		pi.Mutex.RUnlock()
		if stdout == nil {
			return
		}
		select {
		case <-done:
			return
		default:
		}

		n, err := stdout.Read(buffer)
		if n == 0 {
			if err != nil {
				pi.handleProcessExit()
				return
			}
			continue
		}
		decoder.feed(buffer[:n])
		if err != nil {
			pi.handleProcessExit()
			return
		}
	}
}

func (pi *PipeInterface) handleProcessExit() {
	pi.Mutex.Lock()
	pi.Online = false
	enabled := pi.Enabled
	detached := pi.Detached
	command := pi.command
	pi.Mutex.Unlock()
	pi.killProcess()
	debug.Log(debug.DebugInfo, "Subprocess terminated", "name", pi.Name)
	if enabled && !detached && strings.TrimSpace(command) != "" {
		go pi.respawnPipe()
	}
}

func (pi *PipeInterface) handleIOError(err error) {
	debug.Log(debug.DebugError, "Pipe interface write error", "name", pi.Name, "error", err)
	pi.Mutex.Lock()
	pi.Online = false
	pi.Mutex.Unlock()
}

func (pi *PipeInterface) respawnPipe() {
	pi.Mutex.Lock()
	if pi.respawning {
		pi.Mutex.Unlock()
		return
	}
	pi.respawning = true
	pi.Mutex.Unlock()
	defer func() {
		pi.Mutex.Lock()
		pi.respawning = false
		pi.Mutex.Unlock()
	}()

	for {
		pi.Mutex.RLock()
		enabled := pi.Enabled
		detached := pi.Detached
		delay := pi.respawnDelay
		done := pi.done
		pi.Mutex.RUnlock()
		if !enabled || detached {
			return
		}
		select {
		case <-done:
			return
		case <-time.After(delay):
		}
		debug.Log(debug.DebugVerbose, "Attempting to respawn subprocess", "name", pi.Name)
		if err := pi.openPipe(); err != nil {
			debug.Log(debug.DebugError, "Pipe respawn failed", "name", pi.Name, "error", err)
			if pi.panicOnError {
				panic(fmt.Sprintf("pipe interface %s: %v", pi.Name, err))
			}
			continue
		}
		debug.Log(debug.DebugInfo, "Reconnected pipe", "name", pi.Name)
		pi.readLoop()
		return
	}
}

func (pi *PipeInterface) GetConn() net.Conn { return nil }

func (pi *PipeInterface) SendPathRequest([]byte) error { return nil }

func (pi *PipeInterface) SendLinkPacket([]byte, []byte, time.Time) error { return nil }

func (pi *PipeInterface) GetBandwidthAvailable() bool {
	pi.Mutex.RLock()
	defer pi.Mutex.RUnlock()
	return pi.Online && pi.stdin != nil
}

// splitPipeCommand splits a command string on whitespace with quote support.
func splitPipeCommand(command string) ([]string, error) {
	command = strings.TrimSpace(command)
	if command == "" {
		return nil, fmt.Errorf("empty command")
	}
	var (
		args    []string
		cur     strings.Builder
		inQuote bool
		quoteCh byte
		escaped bool
	)
	for i := 0; i < len(command); i++ {
		c := command[i]
		if escaped {
			cur.WriteByte(c)
			escaped = false
			continue
		}
		if inQuote {
			if c == quoteCh {
				inQuote = false
				continue
			}
			if c == '\\' && quoteCh == '"' {
				escaped = true
				continue
			}
			cur.WriteByte(c)
			continue
		}
		switch c {
		case '"', '\'':
			inQuote = true
			quoteCh = c
		case ' ', '\t':
			if cur.Len() > 0 {
				args = append(args, cur.String())
				cur.Reset()
			}
		default:
			cur.WriteByte(c)
		}
	}
	if inQuote {
		return nil, fmt.Errorf("unclosed quote in command")
	}
	if cur.Len() > 0 {
		args = append(args, cur.String())
	}
	if len(args) == 0 {
		return nil, fmt.Errorf("empty command")
	}
	return args, nil
}