| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "log" |
| 7 | "sync" |
| 8 | "time" |
| 9 | |
| 10 | "reasonix/internal/remote/bootstrap" |
| 11 | ) |
| 12 | |
| 13 | type credentialWatchdog struct { |
| 14 | mu sync.Mutex |
| 15 | cancel context.CancelFunc |
| 16 | workspace string |
| 17 | } |
| 18 | |
| 19 | func (w *credentialWatchdog) stop() { |
| 20 | w.mu.Lock() |
| 21 | cancel := w.cancel |
| 22 | w.cancel, w.workspace = nil, "" |
| 23 | w.mu.Unlock() |
| 24 | if cancel != nil { |
| 25 | cancel() |
| 26 | } |
| 27 | } |
| 28 | |
| 29 | // credentialChannelDecision keeps the health policy testable without SSH. |
| 30 | type credentialChannelDecision struct { |
| 31 | HasForward bool |
| 32 | ForwardPort int |
| 33 | HealedPort int |
| 34 | ProbeOK bool |
| 35 | } |
| 36 | |
| 37 | func (d credentialChannelDecision) needsHeal() bool { |
| 38 | return !d.HasForward || !d.ProbeOK || d.HealedPort <= 0 || d.ForwardPort != d.HealedPort |
| 39 | } |
| 40 | |
| 41 | func credentialWatchdogEligibleState(state string) bool { |
| 42 | return state == "connected" || state == "degraded" |
| 43 | } |
| 44 | |
| 45 | func (m *desktopRemoteManager) startCredentialWatchdogIfEnabled(mh *managedHost, hostID, workspace string) { |
| 46 | entry, err := configuredRemoteHost(hostID) |
| 47 | if err == nil && entry.CredentialProxyEnabled() { |
| 48 | m.startCredentialWatchdog(mh, hostID, workspace) |
| 49 | return |
| 50 | } |
| 51 | if mh != nil { |
| 52 | mh.credWatch.stop() |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | func (m *desktopRemoteManager) finishCredentialServe(ctx context.Context, c desktopSSHClient, mh *managedHost, hostID, workspace string, view RemoteServerView, token string, res bootstrap.Result, enabled bool) (RemoteServerView, string, error) { |
| 57 | if !enabled { |
| 58 | return view, token, nil |
| 59 | } |
| 60 | if err := m.healCredentialChannel(ctx, c, mh, hostID, workspace, view.LocalURL, token, res); err != nil { |
| 61 | failed := RemoteServerView{HostID: hostID, Workspace: workspace, State: "error", Error: err.Error()} |
| 62 | m.publishServerIfCurrent(hostID, mh, failed, "", "") |
| 63 | return failed, "", err |
| 64 | } |
| 65 | m.startCredentialWatchdog(mh, hostID, workspace) |
| 66 | return view, token, nil |
| 67 | } |
| 68 | |
| 69 | // startCredentialWatchdog detects reverse-forward drift while a tab is open. |
| 70 | func (m *desktopRemoteManager) startCredentialWatchdog(mh *managedHost, hostID, workspace string) { |
| 71 | if mh == nil || !m.isCurrent(hostID, mh) { |
| 72 | return |
| 73 | } |
| 74 | mh.credWatch.mu.Lock() |
| 75 | mh.credWatch.workspace = workspace |
| 76 | if mh.credWatch.cancel != nil { |
| 77 | mh.credWatch.mu.Unlock() |
| 78 | return |
| 79 | } |
| 80 | parent := mh.ctx |
| 81 | if parent == nil { |
| 82 | parent = context.Background() |
| 83 | } |
| 84 | ctx, cancel := context.WithCancel(parent) |
| 85 | mh.credWatch.cancel = cancel |
| 86 | mh.credWatch.mu.Unlock() |
| 87 | go func() { |
| 88 | ticker := time.NewTicker(3 * time.Second) |
| 89 | defer ticker.Stop() |
| 90 | for { |
| 91 | select { |
| 92 | case <-ctx.Done(): |
| 93 | return |
| 94 | case <-ticker.C: |
| 95 | } |
| 96 | if !m.isCurrent(hostID, mh) { |
| 97 | return |
| 98 | } |
| 99 | m.checkCredentialChannel(ctx, mh, hostID) |
| 100 | } |
| 101 | }() |
| 102 | } |
| 103 | |
| 104 | func (m *desktopRemoteManager) checkCredentialChannel(ctx context.Context, mh *managedHost, hostID string) { |
| 105 | entry, err := configuredRemoteHost(hostID) |
| 106 | if err != nil || !entry.CredentialProxyEnabled() { |
| 107 | mh.credWatch.stop() |
| 108 | return |
| 109 | } |
| 110 | m.mu.Lock() |
| 111 | state := "" |
| 112 | if m.hosts[hostID] == mh { |
| 113 | state = mh.status.State |
| 114 | } |
| 115 | m.mu.Unlock() |
| 116 | // A reconnect with one failed forward is deliberately published as |
| 117 | // degraded while retaining the live SSH client. That is exactly when the |
| 118 | // credential reverse tunnel may need this watchdog's repair path. |
| 119 | if !credentialWatchdogEligibleState(state) || mh.client == nil { |
| 120 | return |
| 121 | } |
| 122 | port, has := credentialForwardPort(mh.client, hostID) |
| 123 | decision := credentialChannelDecision{HasForward: has, ForwardPort: port, HealedPort: int(mh.credPort.Load()), ProbeOK: has && probeReverseTunnel(mh.client, port) == nil} |
| 124 | if decision.needsHeal() { |
| 125 | m.healCredentialChannelWatchdog(ctx, mh, hostID) |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | // healCredentialChannelWatchdog reopens the fast-reuse gate only after the |
| 130 | // forward, remote provider, live Serve providers, and end-to-end probe agree. |
| 131 | func (m *desktopRemoteManager) healCredentialChannelWatchdog(watchCtx context.Context, mh *managedHost, hostID string) { |
| 132 | mh.credWatch.mu.Lock() |
| 133 | workspace := mh.credWatch.workspace |
| 134 | mh.credWatch.mu.Unlock() |
| 135 | if workspace == "" { |
| 136 | return |
| 137 | } |
| 138 | mh.serveMu.Lock() |
| 139 | defer mh.serveMu.Unlock() |
| 140 | if !m.isCurrent(hostID, mh) || mh.client == nil { |
| 141 | return |
| 142 | } |
| 143 | entry, err := configuredRemoteHost(hostID) |
| 144 | if err != nil || !entry.CredentialProxyEnabled() { |
| 145 | return |
| 146 | } |
| 147 | opCtx, opCancel := managedOperationContext(watchCtx, mh) |
| 148 | defer opCancel() |
| 149 | c := mh.client |
| 150 | workspaces := m.trackedCredentialWorkspaces(hostID, workspace) |
| 151 | log.Printf("[remote] credential watchdog: channel broken, re-healing host=%s workspaces=%d", hostID, len(workspaces)) |
| 152 | _ = c.Forwards().Remove("cred-proxy:" + hostID) |
| 153 | healCtx, healCancel := context.WithTimeout(opCtx, credentialProviderHealBudget(len(workspaces))) |
| 154 | err = healTrackedCredentialProviders(healCtx, workspaces, |
| 155 | func(workspace string) (*bootstrap.CredentialProxyOptions, error) { |
| 156 | return m.credentialProxySetup(c, hostID, workspace) |
| 157 | }, |
| 158 | func(ctx context.Context, opts *bootstrap.CredentialProxyOptions) error { |
| 159 | _, healErr := bootstrap.HealCredentialProvider(ctx, c, opts) |
| 160 | return healErr |
| 161 | }, |
| 162 | ) |
| 163 | healCancel() |
| 164 | if err != nil { |
| 165 | log.Printf("[remote] credential watchdog: config heal FAILED host=%s err=%v", hostID, err) |
| 166 | return |
| 167 | } |
| 168 | port, has := credentialForwardPort(c, hostID) |
| 169 | if !has { |
| 170 | log.Printf("[remote] credential watchdog: forward missing after setup host=%s", hostID) |
| 171 | return |
| 172 | } |
| 173 | if err := probeReverseTunnel(c, port); err != nil { |
| 174 | log.Printf("[remote] credential watchdog: probe still FAILED host=%s port=%d err=%v", hostID, port, err) |
| 175 | return |
| 176 | } |
| 177 | if !m.isCurrent(hostID, mh) { |
| 178 | return |
| 179 | } |
| 180 | reloadCtx, reloadCancel := context.WithTimeout(opCtx, credentialProviderReloadBudget(len(workspaces))) |
| 181 | reloadOK := m.reloadServeProviders(reloadCtx, mh, hostID, workspace, "", "") |
| 182 | reloadCancel() |
| 183 | if !reloadOK { |
| 184 | log.Printf("[remote] credential watchdog: provider reload FAILED host=%s", hostID) |
| 185 | return |
| 186 | } |
| 187 | if !m.isCurrent(hostID, mh) { |
| 188 | return |
| 189 | } |
| 190 | mh.credPort.Store(int64(port)) |
| 191 | log.Printf("[remote] credential watchdog: channel re-healed host=%s port=%d", hostID, port) |
| 192 | } |
| 193 | |
| 194 | func credentialProviderReloadBudget(targets int) time.Duration { |
| 195 | if targets < 1 { |
| 196 | targets = 1 |
| 197 | } |
| 198 | return time.Duration(targets) * remoteProviderReloadTimeout |
| 199 | } |
| 200 | |
| 201 | func credentialProviderHealBudget(targets int) time.Duration { |
| 202 | if targets < 1 { |
| 203 | targets = 1 |
| 204 | } |
| 205 | return time.Duration(targets) * 30 * time.Second |
| 206 | } |
| 207 | |
| 208 | func healTrackedCredentialProviders(ctx context.Context, workspaces []string, setup func(string) (*bootstrap.CredentialProxyOptions, error), heal func(context.Context, *bootstrap.CredentialProxyOptions) error) error { |
| 209 | for _, workspace := range workspaces { |
| 210 | opts, err := setup(workspace) |
| 211 | if err != nil { |
| 212 | return fmt.Errorf("workspace %q setup: %w", workspace, err) |
| 213 | } |
| 214 | if err := heal(ctx, opts); err != nil { |
| 215 | return fmt.Errorf("workspace %q heal: %w", workspace, err) |
| 216 | } |
| 217 | } |
| 218 | return nil |
| 219 | } |
| 220 | |
| 221 | func healCredentialConfigsBeforeReload(ctx context.Context, workspaces []string, setup func(string) (*bootstrap.CredentialProxyOptions, error), heal func(context.Context, *bootstrap.CredentialProxyOptions) error, reload func() bool) error { |
| 222 | if err := healTrackedCredentialProviders(ctx, workspaces, setup, heal); err != nil { |
| 223 | return fmt.Errorf("credential proxy: heal tracked provider configs: %w", err) |
| 224 | } |
| 225 | if reload == nil || !reload() { |
| 226 | return fmt.Errorf("credential proxy: serve providers could not reload") |
| 227 | } |
| 228 | return nil |
| 229 | } |
| 230 |