diff --git a/pkg/cloudproxy/agent/ssh/client.go b/pkg/cloudproxy/agent/ssh/client.go index 75e4560ea0..4b9143e3eb 100644 --- a/pkg/cloudproxy/agent/ssh/client.go +++ b/pkg/cloudproxy/agent/ssh/client.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "net" + "strconv" "sync" "time" @@ -150,9 +151,31 @@ func (c *Client) Start(ctx context.Context) { sshClientC := make(chan *ssh.Client) var sshClient *ssh.Client go c.runClientState(ctx, sshClientC) + for { select { - case sshClient = <-sshClientC: + case sshc := <-sshClientC: + conn := sshc.Conn + localAddr := conn.LocalAddr() + localAddrStr := localAddr.String() + addr, portStr, err := net.SplitHostPort(localAddrStr) + if err != nil { + log.Errorf("split host port of ssh client local addr: %v", err) + sshc.Close() + break + } + port, err := strconv.ParseUint(portStr, 10, 16) + if err != nil { + log.Errorf("parse ssh client local port: %v", err) + sshc.Close() + break + } + if v := c.localForwards.get(int(port), addr); v != nil { + log.Errorf("ssh client local port %d collides with local forward: %#v", port, v) + sshc.Close() + break + } + sshClient = sshc case req := <-c.lfc: if sshClient != nil { c.localForward(ctx, sshClient, req) @@ -178,7 +201,7 @@ func (c *Client) Start(ctx context.Context) { } } -func (c *Client) runClientState(ctx context.Context, sshClientC chan *ssh.Client) { +func (c *Client) runClientState(ctx context.Context, sshClientC chan<- *ssh.Client) { for { select { case <-ctx.Done(): @@ -203,12 +226,6 @@ func (c *Client) runClientState(ctx context.Context, sshClientC chan *ssh.Client func() { defer sshc.Conn.Close() - select { - case sshClientC <- sshc: - case <-ctx.Done(): - return - } - closeC := make(chan struct{}) go func() { defer close(closeC) @@ -219,6 +236,13 @@ func (c *Client) runClientState(ctx context.Context, sshClientC chan *ssh.Client } }() + select { + case sshClientC <- sshc: + case <-closeC: + case <-ctx.Done(): + return + } + select { case <-closeC: case <-ctx.Done(): diff --git a/pkg/cloudproxy/agent/worker/worker.go b/pkg/cloudproxy/agent/worker/worker.go index 9ef20731a1..ea663f6d16 100644 --- a/pkg/cloudproxy/agent/worker/worker.go +++ b/pkg/cloudproxy/agent/worker/worker.go @@ -180,14 +180,24 @@ func (w *Worker) Start(ctx context.Context) { } go w.apih.Start(ctx) - var mss *agentmodels.ModelSets + const tickDur = 11 * time.Second + var ( + mss *agentmodels.ModelSets + tick = time.NewTicker(tickDur) + ) for { select { case imss := <-w.apih.ModelSets(): log.Infof("agent: got new data from api helper") mss = imss.(*agentmodels.ModelSets) if err := w.run(ctx, mss); err != nil { - log.Errorf("agent: %v", err) + log.Errorf("agent run: %v", err) + } + case <-tick.C: + if mss != nil { + if err := w.run(ctx, mss); err != nil { + log.Errorf("agent refresh run: %v", err) + } } case <-ctx.Done(): return diff --git a/pkg/compute/models/guest_sshable.go b/pkg/compute/models/guest_sshable.go index 4ce414fb59..6db66db36e 100644 --- a/pkg/compute/models/guest_sshable.go +++ b/pkg/compute/models/guest_sshable.go @@ -213,10 +213,11 @@ func (guest *SGuest) sshableTryEach( tmo := time.NewTimer(13 * time.Second) tick := time.NewTicker(3 * time.Second) + out: for { select { case <-tmo.C: - break + break out case <-tick.C: if ok := guest.sshableTryForward(ctx, tryData, &fwd); ok { return nil