ham/protocol/aprs/aprsis/proxy.go

package aprsis

import (
	"bufio"
	"bytes"
	"fmt"
	"math/rand/v2"
	"net"
	"strings"
	"sync"
	"time"

	"git.maze.io/go/ham/protocol"
	"git.maze.io/go/ham/radio"
	"github.com/sirupsen/logrus"
)

const (
	DefaultListenAddr = ":14580"
	DefaultServerAddr = "rotate.aprs2.net:14580"
)

// OnProxyClientFunc callback.
type OnProxyClientFunc func(callsign string, client *ProxyClient)

type Proxy struct {
	// Logger used for the proxy.
	Logger *logrus.Logger

	// Filter is an APRS range filter.
	Filter string

	// OnClient gets called when a new client is logged in through the proxy.
	OnClient OnProxyClientFunc

	server    string
	listen    net.Listener
	clientsMu sync.RWMutex
	clients   map[uint64]*ProxyClient
}

func NewProxy(listen, server string) (*Proxy, error) {
	if _, err := net.ResolveTCPAddr("tcp", server); err != nil {
		return nil, fmt.Errorf("aprsis: error resolving %q: %v", server, err)
	}

	listenAddr, err := net.ResolveTCPAddr("tcp", listen)
	if err != nil {
		return nil, fmt.Errorf("aprsis: error listening on %s: %v", listen, err)
	}

	listener, err := net.ListenTCP("tcp", listenAddr)
	if err != nil {
		return nil, fmt.Errorf("aprsis: error listening on %s: %v", listen, err)
	}

	proxy := &Proxy{
		Logger:  logrus.New(),
		server:  server,
		listen:  listener,
		clients: make(map[uint64]*ProxyClient),
	}
	go proxy.accept()

	return proxy, nil
}

func (proxy *Proxy) Close() error {
	return proxy.listen.Close()
}

func (proxy *Proxy) accept() {
	for {
		conn, err := proxy.listen.Accept()
		if err != nil {
			proxy.Logger.Errorf("aprs-is proxy: error accepting client: %v", err)
			continue
		}

		client := NewProxyClient(conn)
		client.filter = proxy.Filter
		client.callback = func(callsign string, client *ProxyClient) {
			if proxy.OnClient != nil {
				proxy.OnClient(callsign, client)
			}
		}
		go client.Run(proxy)
	}
}

func (proxy *Proxy) addClient(client *ProxyClient) {
	proxy.clientsMu.Lock()
	proxy.clients[client.id] = client
	proxy.clientsMu.Unlock()
}

func (proxy *Proxy) removeClient(client *ProxyClient) {
	proxy.clientsMu.Lock()
	delete(proxy.clients, client.id)
	proxy.clientsMu.Unlock()
}

type ProxyClient struct {
	net.Conn
	logger   *logrus.Logger
	id       uint64
	filter   string
	myCall   string
	packets  chan *protocol.Packet
	callback OnProxyClientFunc
}

func NewProxyClient(conn net.Conn) *ProxyClient {
	return &ProxyClient{
		Conn:   conn,
		id:     uint64(rand.Int64()),
		logger: logrus.New(),
	}
}

func (client *ProxyClient) Close() error {
	if client.packets != nil {
		close(client.packets)
		client.packets = nil
	}
	return client.Conn.Close()
}

func (client *ProxyClient) Run(proxy *Proxy) {
	proxy.addClient(client)
	defer func() {
		proxy.removeClient(client)
		_ = client.Close()
	}()

	host, _, _ := net.SplitHostPort(client.RemoteAddr().String())
	proxy.Logger.Infof("aprs-is proxy[%s]: new connection", host)

	server, err := net.Dial("tcp", proxy.server)
	if err != nil {
		proxy.Logger.Warnf("aprs-is proxy[%s]: can't connecto to APRS-IS server %s: %v", host, proxy.server, err)
		return
	}
	defer func() { _ = server.Close() }()

	var (
		wait sync.WaitGroup
		call string
	)
	wait.Go(func() { client.copy(client, server, host, "->", &call) })
	wait.Go(func() { client.copy(server, client, host, "<-", nil) })
	wait.Wait()
}

func (client *ProxyClient) copy(dst, src net.Conn, host, dir string, call *string) {
	defer func() {
		if tcp, ok := dst.(*net.TCPConn); ok {
			_ = tcp.CloseWrite()
		} else {
			_ = dst.Close()
		}
	}()

	reader := bufio.NewReader(src)
	for {
		line, err := reader.ReadBytes('\n')
		if err != nil {
			client.logger.Warnf("aprs-is proxy[%s]: %s read error: %v", host, src.RemoteAddr(), err)
			return
		}

		// proxy to remote unaltered
		if len(line) > 0 {
			if _, err = dst.Write(line); err != nil {
				client.logger.Warnf("aprs-is proxy[%s]: %s write error: %v", host, dst.RemoteAddr(), err)
				return
			}
		}

		// parse line
		line = bytes.TrimRight(line, "\r\n")
		if len(line) > 0 {
			client.logger.Tracef("aprs-is proxy[%s]: %s %s", host, dir, string(line))

			if strings.HasPrefix(string(line), "# logresp ") {
				// server responds to client login
				part := strings.SplitN(string(line), " ", 5)
				if len(part) > 4 && part[3] == "verified," {
					// Keep track of our call.
					client.myCall = part[2]
					if call != nil {
						*call = part[2]
					}
					client.logger.Infof("aprs-is proxy[%s]: logged in as %s", host, *call)

					// Invoke OnClient callback.
					client.callback(client.myCall, client)

					// Send optional range filter to the APRS-IS server.
					if client.filter != "" {
						client.logger.Tracef("aprs-is proxy[%s]: %s filter %s", host, dir, client.filter)
						if _, err = fmt.Fprintf(src, "filter %s\r\n", client.filter); err != nil {
							client.logger.Warnf("aprs-is proxy[%s]: %s write error: %v", host, src.RemoteAddr(), err)
							return
						}
					}
				}
			}

			if !isCommand(line) {
				client.handleRawPacket(line)
			}
		}
	}
}

func (client *ProxyClient) Info() *radio.Info {
	// We have very little information actually, but here we go:
	return &radio.Info{
		Name: client.myCall,
	}
}

func (client *ProxyClient) RawPackets() <-chan *protocol.Packet {
	if client.packets == nil {
		client.packets = make(chan *protocol.Packet, 16)
	}
	return client.packets
}

func (client *ProxyClient) handleRawPacket(data []byte) {
	if client.packets == nil {
		return
	}

	select {
	case client.packets <- &protocol.Packet{
		Time:     time.Now().UTC(),
		Protocol: protocol.APRS,
		Raw:      data,
	}:
	default:
		client.logger.Warn("aprs-is proxy: raw packet channel full, dropping packet")
	}
}

func isCommand(line []byte) bool {
	if len(line) == 0 {
		return true
	}

	if line[0] == '#' {
		return true
	}

	if i := bytes.IndexByte(line, ' '); i > -1 {
		switch strings.ToLower(string(line[:i])) {
		case "user", "filter":
			return true
		}
	}

	return false
}