johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package guestagent

import (
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"os"
	"strings"
	"sync"

	"filippo.io/age"
	"github.com/pion/ice/v4"
	"github.com/pion/webrtc/v4"
)

type Agent struct {
	cfg Config

	log *slog.Logger

	identity *age.X25519Identity

	certificate webrtc.Certificate

	fingerprint string

	api *webrtc.API

	mux ice.UDPMux

	tokens *tokenSource

	poster *poster

	ctx context.Context

	wg sync.WaitGroup

	mu sync.Mutex

	sessions map[string]*session

	closed bool

	cursor int64

	seen map[string]bool

	requested bool

	applied map[string]bool

	reportedDID string

	sharedGID int

	shared bool
}

func New(cfg Config, log *slog.Logger) (*Agent, error) {
	identity, err := age.GenerateX25519Identity()
	if err != nil {
		return nil, fmt.Errorf("guestagent: generating the age identity: %w", err)
	}
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		return nil, fmt.Errorf("guestagent: generating the DTLS key: %w", err)
	}
	certificate, err := webrtc.GenerateCertificate(key)
	if err != nil {
		return nil, fmt.Errorf("guestagent: generating the DTLS certificate: %w", err)
	}
	fingerprint, err := Fingerprint(*certificate)
	if err != nil {
		return nil, err
	}
	settings := webrtc.SettingEngine{LoggerFactory: pionLogs{log: log}}
	settings.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeUDP4, webrtc.NetworkTypeUDP6})
	mux, err := forwardedPort(&settings, cfg.Bidder.WebRTC, log)
	if err != nil {
		return nil, err
	}
	tokens := newTokenSource(cfg.LoginHelper, cfg.Bidder.Audience, log)
	return &Agent{
		cfg:         cfg,
		log:         log,
		identity:    identity,
		certificate: *certificate,
		fingerprint: fingerprint,
		api:         webrtc.NewAPI(webrtc.WithSettingEngine(settings)),
		mux:         mux,
		tokens:      tokens,
		poster: &poster{
			url:    cfg.Bidder.Base + EventsPath,
			accept: cfg.Accept,
			tokens: tokens,
			client: &http.Client{Transport: &http.Transport{Proxy: nil}},
			log:    log,
		},
		sessions: map[string]*session{},
		seen:     map[string]bool{},
		applied:  map[string]bool{},
	}, nil
}

func forwardedPort(settings *webrtc.SettingEngine, w *WebRTCPort, log *slog.Logger) (ice.UDPMux, error) {
	if w == nil {
		return nil, nil
	}
	conn, err := net.ListenUDP("udp", &net.UDPAddr{Port: w.UDPPort})
	if err != nil {
		log.Warn("the forwarded webrtc UDP port does not bind; host candidates take ephemeral ports", "port", w.UDPPort, "err", err.Error())
		return nil, nil
	}
	mux := ice.NewUDPMuxDefault(ice.UDPMuxParams{UDPConn: conn, Logger: pionLogs{log: log}.NewLogger("udpmux")})
	settings.SetICEUDPMux(mux)
	rule := webrtc.ICEAddressRewriteRule{External: w.HostAddresses, AsCandidateType: webrtc.ICECandidateTypeHost, Mode: webrtc.ICEAddressRewriteReplace}
	if err := settings.SetICEAddressRewriteRules(rule); err != nil {
		_ = mux.Close()
		return nil, fmt.Errorf("guestagent: the host candidate rewrite to the bidder host's LAN addresses: %w", err)
	}
	return mux, nil
}

func Fingerprint(certificate webrtc.Certificate) (string, error) {
	fingerprints, err := certificate.GetFingerprints()
	if err != nil {
		return "", fmt.Errorf("guestagent: the DTLS certificate fingerprint: %w", err)
	}
	for _, f := range fingerprints {
		if f.Algorithm == "sha-256" {
			return f.Algorithm + " " + strings.ToUpper(f.Value), nil
		}
	}
	return "", fmt.Errorf("guestagent: the DTLS certificate has no sha-256 fingerprint")
}

func (a *Agent) Fingerprint() string { return a.fingerprint }

func (a *Agent) Run(ctx context.Context, ready func()) error {
	if err := os.MkdirAll(a.cfg.StateDir, 0o700); err != nil {
		return fmt.Errorf("%w: the state dir: %v", ErrConfig, err)
	}
	a.ctx = ctx
	a.loadCursor()
	tags := make([]string, 0, len(a.cfg.Tunnels))
	for _, t := range a.cfg.Tunnels {
		tags = append(tags, t.Tag)
	}
	a.log.Info("the guest agent starts", "tags", tags, "sealedSecrets", a.cfg.Secrets != nil, "agentUnit", a.cfg.AgentUnit, "association", a.cfg.Association != nil, "base", a.cfg.Bidder.Base, "fingerprint", a.fingerprint, "cursor", a.cursor,
		"forwardedUDPPort", a.mux != nil)
	for _, t := range a.cfg.Tunnels {
		a.wg.Go(func() { a.announce(ctx, t.Tag) })
	}
	if a.cfg.Secrets != nil {
		a.wg.Go(func() { a.runSecrets(ctx) })
	}
	if a.cfg.Association != nil {
		a.wg.Go(func() { a.runAssociation(ctx) })
	}
	if ready != nil {
		ready()
	}
	a.subscribe(ctx)
	a.mu.Lock()
	a.closed = true
	open := make([]*session, 0, len(a.sessions))
	for _, s := range a.sessions {
		open = append(open, s)
	}
	a.mu.Unlock()
	for _, s := range open {
		s.close("the agent is stopping")
	}
	a.wg.Wait()
	if a.mux != nil {
		_ = a.mux.Close()
	}
	a.log.Info("the guest agent stops")
	return nil
}