fix: fall back to enroll token when pilot tunnel JWT is rejected (#1566)

* fix: fall back to enroll token when pilot tunnel JWT is rejected

On HTTP 401/404 upgrade rejection, delete stale pilot.jwt and retry with SENCHO_ENROLL_TOKEN.

* test: import pilot agent module after DATA_DIR is set in fallback test

The auth-fallback test statically imported pilot/agent, which freezes its
pilot.jwt path from DATA_DIR at module load, before setupTestDb redirects
DATA_DIR to a writable temp dir. On the Linux CI runner the path resolved
to a non-writable /app/data, so persistToken silently failed and the
round-trip assertion read null. Import the module dynamically in beforeAll
after setupTestDb, matching the sibling unit test.

* fix: remove unexpected-response listener that blocked pilot reconnect

The ws library skips abortHandshake when an unexpected-response listener
exists, so error and close never fire and the agent hangs in CONNECTING.
Detect auth rejection via the abortHandshake error message instead; the
close handler already performs enroll-token fallback and reconnect.

backend/src/__tests__/pilot-agent-auth-fallback-unit.test.ts

@@ -0,0 +1,129 @@
+/**
+ * Unit tests for PilotAgent auth fallback event handling using a stub WebSocket.
+ */
+import { describe, it, expect, vi, beforeEach, afterEach, beforeAll, afterAll } from 'vitest';
+import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
+
+const { wsInstances, mockAttachSwitchboard } = vi.hoisted(() => ({
+    wsInstances: [] as Array<{
+        emit: (event: string, ...args: unknown[]) => boolean;
+        readyState: number;
+        close: ReturnType<typeof vi.fn>;
+    }>,
+    mockAttachSwitchboard: vi.fn(() => ({
+        handleJsonFrame: vi.fn(() => false),
+        handleBinaryFrame: vi.fn(() => false),
+        cleanup: vi.fn(),
+        tcpStreamCount: vi.fn(() => 0),
+        openReverseStream: vi.fn(() => null),
+    })),
+}));
+
+vi.mock('../mesh/tcpStreamSwitchboard', () => ({
+    attachTcpStreamSwitchboard: mockAttachSwitchboard,
+    resolveByComposeLabels: vi.fn(),
+}));
+
+vi.mock('ws', () => {
+    // eslint-disable-next-line @typescript-eslint/no-require-imports
+    const { EventEmitter } = require('events') as typeof import('events');
+    class MockWebSocket extends EventEmitter {
+        readyState = 0;
+        close = vi.fn();
+        constructor(..._args: unknown[]) {
+            super();
+            wsInstances.push(this);
+        }
+    }
+    return { default: MockWebSocket };
+});
+
+let tmpDir: string;
+let PilotAgent: typeof import('../pilot/agent').PilotAgent;
+let readPersistedToken: typeof import('../pilot/agent').readPersistedToken;
+let persistToken: typeof import('../pilot/agent').persistToken;
+let clearPersistedToken: typeof import('../pilot/agent').clearPersistedToken;
+
+beforeAll(async () => {
+    tmpDir = await setupTestDb();
+    ({ PilotAgent, readPersistedToken, persistToken, clearPersistedToken } = await import('../pilot/agent'));
+});
+
+afterAll(() => {
+    cleanupTestDb(tmpDir);
+});
+
+describe('PilotAgent auth fallback (stub WebSocket)', () => {
+    beforeEach(() => {
+        wsInstances.length = 0;
+        vi.clearAllMocks();
+        vi.spyOn(console, 'log').mockImplementation(() => { /* swallow */ });
+        vi.spyOn(console, 'warn').mockImplementation(() => { /* swallow */ });
+    });
+
+    afterEach(() => {
+        vi.restoreAllMocks();
+        clearPersistedToken();
+    });
+
+    it('swaps to the enroll token and clears pilot.jwt after HTTP 401 on upgrade', () => {
+        persistToken('stale-on-disk');
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'stale-token',
+            enrollToken: 'fresh-enroll-token',
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+
+        const firstWs = wsInstances[0]!;
+        // Real ws (no unexpected-response listener): abortHandshake emits error then close.
+        firstWs.emit('error', new Error('Unexpected server response: 401'));
+        firstWs.emit('close', 1006, Buffer.from(''));
+
+        expect(readPersistedToken()).toBeNull();
+        expect((agent as unknown as { token: string }).token).toBe('fresh-enroll-token');
+        // scheduleReconnect doubles backoff after scheduling the imminent retry.
+        expect((agent as unknown as { backoff: number }).backoff).toBe(2_000);
+    });
+
+    it('swaps to the enroll token after HTTP 404 on upgrade', () => {
+        persistToken('stale-on-disk');
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'stale-token',
+            enrollToken: 'fresh-enroll-token',
+            enrolling: false,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+
+        const firstWs = wsInstances[0]!;
+        firstWs.emit('error', new Error('Unexpected server response: 404'));
+        firstWs.emit('close', 1006, Buffer.from(''));
+
+        expect(readPersistedToken()).toBeNull();
+        expect((agent as unknown as { token: string }).token).toBe('fresh-enroll-token');
+    });
+
+    it('does not swap when already connecting with the enroll token', () => {
+        const agent = new PilotAgent({
+            primaryUrl: 'http://primary.invalid',
+            loopbackPort: 1,
+            initialToken: 'only-enroll-token',
+            enrollToken: 'only-enroll-token',
+            enrolling: true,
+        });
+
+        (agent as unknown as { connect: () => void }).connect();
+
+        const firstWs = wsInstances[0]!;
+        firstWs.emit('error', new Error('Unexpected server response: 401'));
+        firstWs.emit('close', 1006, Buffer.from(''));
+
+        expect((agent as unknown as { token: string }).token).toBe('only-enroll-token');
+    });
+});

backend/src/__tests__/pilot-agent-auth-fallback.test.ts

@@ -0,0 +1,150 @@
+/**
+ * Regression tests for pilot agent auth fallback when a persisted tunnel
+ * token is rejected at WebSocket upgrade (401 invalid JWT, 404 unknown node).
+ *
+ * Without fallback the agent reconnects forever with the same stale
+ * pilot.jwt credential even when SENCHO_ENROLL_TOKEN carries a fresh
+ * enrollment JWT.
+ */
+import { describe, it, expect, beforeAll, afterAll, afterEach } from 'vitest';
+import http from 'http';
+import jwt from 'jsonwebtoken';
+import { WebSocketServer } from 'ws';
+import { setupTestDb, cleanupTestDb, TEST_JWT_SECRET } from './helpers/setupTestDb';
+import { attachUpgrade } from '../websocket/upgradeHandler';
+import { PilotTunnelManager } from '../services/PilotTunnelManager';
+import { WebSocket } from 'ws';
+
+let tmpDir: string;
+let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
+
+// agent.ts freezes its pilot.jwt path from DATA_DIR at module load, so it must
+// be imported only after setupTestDb() points DATA_DIR at the writable tmp dir.
+let readPersistedToken: typeof import('../pilot/agent').readPersistedToken;
+let persistToken: typeof import('../pilot/agent').persistToken;
+let clearPersistedToken: typeof import('../pilot/agent').clearPersistedToken;
+
+let server: http.Server;
+let port: number;
+let pilotTunnelWss: WebSocketServer;
+let mainWss: WebSocketServer;
+let nodeId: number;
+
+beforeAll(async () => {
+    tmpDir = await setupTestDb();
+    ({ DatabaseService } = await import('../services/DatabaseService'));
+    ({ readPersistedToken, persistToken, clearPersistedToken } = await import('../pilot/agent'));
+
+    server = http.createServer();
+    mainWss = new WebSocketServer({ noServer: true });
+    pilotTunnelWss = new WebSocketServer({ noServer: true });
+    attachUpgrade(server, { wss: mainWss, pilotTunnelWss });
+
+    await new Promise<void>((resolve, reject) => {
+        server.once('error', reject);
+        server.listen(0, '127.0.0.1', () => {
+            const addr = server.address();
+            if (!addr || typeof addr === 'string') {
+                reject(new Error('listen returned unexpected address'));
+                return;
+            }
+            port = addr.port;
+            resolve();
+        });
+    });
+
+    nodeId = DatabaseService.getInstance().addNode({
+        name: `pilot-auth-fallback-${Date.now()}`,
+        type: 'remote',
+        mode: 'pilot_agent',
+        compose_dir: '/tmp/x',
+        is_default: false,
+        api_url: '',
+        api_token: '',
+    });
+});
+
+afterAll(async () => {
+    const mgr = PilotTunnelManager.getInstance();
+    mgr.closeTunnel(nodeId);
+    mgr.removeAllListeners('tunnel-up');
+    mgr.removeAllListeners('tunnel-down');
+    pilotTunnelWss.close();
+    mainWss.close();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+    cleanupTestDb(tmpDir);
+});
+
+afterEach(() => {
+    PilotTunnelManager.getInstance().closeTunnel(nodeId);
+    clearPersistedToken();
+});
+
+function mintStaleTunnelTokenWrongSecret(): string {
+    return jwt.sign(
+        { scope: 'pilot_tunnel', nodeId },
+        'wrong-secret-not-the-control-instance',
+        { expiresIn: '365d' },
+    );
+}
+
+function mintStaleTunnelTokenWrongNode(): string {
+    return jwt.sign(
+        { scope: 'pilot_tunnel', nodeId: 99_999_999 },
+        TEST_JWT_SECRET,
+        { expiresIn: '365d' },
+    );
+}
+
+describe('clearPersistedToken', () => {
+    it('removes an existing pilot.jwt file', () => {
+        persistToken('stale-token');
+        expect(readPersistedToken()).toBe('stale-token');
+        clearPersistedToken();
+        expect(readPersistedToken()).toBeNull();
+    });
+
+    it('does not throw when the file is already absent', () => {
+        clearPersistedToken();
+        expect(() => clearPersistedToken()).not.toThrow();
+        expect(readPersistedToken()).toBeNull();
+    });
+});
+
+describe('pilot tunnel upgrade rejection (in-process integration)', () => {
+    it('rejects a stale tunnel JWT signed with the wrong secret at upgrade', async () => {
+        const staleToken = mintStaleTunnelTokenWrongSecret();
+        const ws = new WebSocket(`ws://127.0.0.1:${port}/api/pilot/tunnel`, {
+            headers: {
+                Authorization: `Bearer ${staleToken}`,
+                'x-sencho-agent-version': 'auth-fallback-test/1.0',
+            },
+        });
+        const result = await new Promise<{ status?: number }>((resolve) => {
+            ws.on('unexpected-response', (_req, res) => {
+                resolve({ status: res.statusCode });
+                res.destroy();
+            });
+            ws.on('error', () => { /* close follows */ });
+        });
+        expect(result.status).toBe(401);
+    });
+
+    it('rejects a tunnel JWT for an unknown node with HTTP 404 at upgrade', async () => {
+        const staleToken = mintStaleTunnelTokenWrongNode();
+        const ws = new WebSocket(`ws://127.0.0.1:${port}/api/pilot/tunnel`, {
+            headers: {
+                Authorization: `Bearer ${staleToken}`,
+                'x-sencho-agent-version': 'auth-fallback-test/1.0',
+            },
+        });
+        const result = await new Promise<{ status?: number }>((resolve) => {
+            ws.on('unexpected-response', (_req, res) => {
+                resolve({ status: res.statusCode });
+                res.destroy();
+            });
+            ws.on('error', () => { /* close follows */ });
+        });
+        expect(result.status).toBe(404);
+    });
+});

backend/src/__tests__/pilot-agent-fs-errors.test.ts

@@ -19,11 +19,12 @@
  */
 import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
 
-const { mockReadFileSync, mockWriteFileSync, mockExistsSync, mockMkdirSync } = vi.hoisted(() => ({
+const { mockReadFileSync, mockWriteFileSync, mockExistsSync, mockMkdirSync, mockUnlinkSync } = vi.hoisted(() => ({
     mockReadFileSync: vi.fn(),
     mockWriteFileSync: vi.fn(),
     mockExistsSync: vi.fn(),
     mockMkdirSync: vi.fn(),
+    mockUnlinkSync: vi.fn(),
 }));
 
 vi.mock('fs', () => {
@@ -32,13 +33,14 @@ vi.mock('fs', () => {
         writeFileSync: mockWriteFileSync,
         existsSync: mockExistsSync,
         mkdirSync: mockMkdirSync,
+        unlinkSync: mockUnlinkSync,
     };
     return { ...mock, default: mock };
 });
 
 // agent.ts is imported AFTER vi.mock so the mock is in place when the
 // module's top-level fs import resolves.
-import { readPersistedToken, persistToken } from '../pilot/agent';
+import { readPersistedToken, persistToken, clearPersistedToken } from '../pilot/agent';
 
 let errorSpy: ReturnType<typeof vi.spyOn>;
 let warnSpy: ReturnType<typeof vi.spyOn>;
@@ -167,3 +169,25 @@ describe('persistToken', () => {
         expect(() => persistToken('test-token')).not.toThrow();
     });
 });
+
+describe('clearPersistedToken', () => {
+    it('unlinks the token file on the happy path', () => {
+        clearPersistedToken();
+        expect(mockUnlinkSync).toHaveBeenCalledWith(expect.stringContaining('pilot.jwt'));
+        expect(warnSpy).not.toHaveBeenCalled();
+    });
+
+    it('does not warn on ENOENT (file already absent)', () => {
+        mockUnlinkSync.mockImplementationOnce(() => { throw fsError('ENOENT', 'no such file'); });
+        clearPersistedToken();
+        expect(warnSpy).not.toHaveBeenCalled();
+    });
+
+    it('warns on EACCES (read-only volume)', () => {
+        mockUnlinkSync.mockImplementationOnce(() => { throw fsError('EACCES', 'permission denied'); });
+        clearPersistedToken();
+        expect(warnSpy).toHaveBeenCalledOnce();
+        expect(String(warnSpy.mock.calls[0][0])).toContain('Failed to remove persisted tunnel token');
+        expect(String(warnSpy.mock.calls[0][0])).toContain('EACCES');
+    });
+});

backend/src/__tests__/pilot-agent-loopback-auth.test.ts

@@ -33,6 +33,7 @@ describe('PilotAgent loopback auth injection', () => {
       primaryUrl: 'http://primary.invalid',
       loopbackPort: 1,
       initialToken: 'irrelevant-for-this-test',
+      enrollToken: null,
       enrolling: false,
     });
     mintHeader = (agent as unknown as { getLoopbackAuthHeader: () => string | null }).getLoopbackAuthHeader.bind(agent);

backend/src/pilot/agent.ts

@@ -61,6 +61,7 @@ export function startPilotAgent(loopbackPort: number): void {
         primaryUrl,
         loopbackPort,
         initialToken: persistedToken || enrollToken!,
+        enrollToken: enrollToken ?? null,
         enrolling: !persistedToken,
     });
     // Register the agent as MeshService's reverse dialer so outbound
@@ -80,13 +81,18 @@ interface AgentOptions {
     primaryUrl: string;
     loopbackPort: number;
     initialToken: string;
+    enrollToken: string | null;
     enrolling: boolean;
 }
 
 export class PilotAgent {
     private readonly options: AgentOptions;
     private token: string;
+    /** Fallback credential from SENCHO_ENROLL_TOKEN when the persisted token is rejected. */
+    private readonly enrollToken: string | null;
     private backoff = RECONNECT_MIN_MS;
+    /** Set when the WS upgrade is rejected with 401 or 404 before the handshake completes. */
+    private upgradeRejected = false;
     private ws: WebSocket | null = null;
     private pingTimer?: NodeJS.Timeout;
     private reconnectTimer?: NodeJS.Timeout;
@@ -113,6 +119,7 @@ export class PilotAgent {
     constructor(options: AgentOptions) {
         this.options = options;
         this.token = options.initialToken;
+        this.enrollToken = options.enrollToken;
         this.agentVersion = getSenchoVersion() || '0.0.0';
         this.customCa = readPilotCaBundle();
     }
@@ -170,6 +177,8 @@ export class PilotAgent {
     private connect(): void {
         if (this.shuttingDown) return;
 
+        this.upgradeRejected = false;
+
         const wsUrl = httpUrlToWs(this.options.primaryUrl) + '/api/pilot/tunnel';
         const ws = new WebSocket(wsUrl, {
             headers: {
@@ -187,6 +196,22 @@ export class PilotAgent {
             ...(this.customCa ? { ca: this.customCa } : {}),
         });
         this.ws = ws;
+        let opened = false;
+        let lastHandshakeError: string | null = null;
+
+        // Do NOT register an 'unexpected-response' listener here. The ws library
+        // skips abortHandshake (and therefore never emits 'error' or 'close')
+        // when a listener exists for that event. Auth rejection is detected via
+        // the error message abortHandshake emits: "Unexpected server response: N".
+        ws.on('error', (err) => {
+            console.warn('[Pilot] Tunnel error:', err.message);
+            lastHandshakeError = err.message;
+            if (/Unexpected server response: (401|404)/.test(err.message)) {
+                this.upgradeRejected = true;
+            }
+            // 'close' will follow; reconnect is scheduled there.
+        });
+
         this.switchboard = attachTcpStreamSwitchboard({
             ws,
             resolveTarget: resolveByComposeLabels,
@@ -195,6 +220,7 @@ export class PilotAgent {
         });
 
         ws.on('open', () => {
+            opened = true;
             // Backoff intentionally NOT reset here: a TCP-level connect that
             // immediately fails the protocol handshake (incompatible version,
             // bad token consumed at upgrade) would otherwise reset the
@@ -222,12 +248,19 @@ export class PilotAgent {
         ws.on('close', (code, reason) => {
             console.log('[Pilot] Tunnel closed:', code, reason?.toString?.() ?? '');
             this.cleanupAfterDisconnect();
+
+            const authRejected = this.upgradeRejected
+                || (!opened && lastHandshakeError != null && /Unexpected server response: (401|404)/.test(lastHandshakeError));
+            if (authRejected && this.enrollToken && this.token !== this.enrollToken) {
+                clearPersistedToken();
+                console.log('[Pilot] Persisted token rejected; falling back to enroll token');
+                this.token = this.enrollToken;
+                this.backoff = RECONNECT_MIN_MS;
+            }
+            this.upgradeRejected = false;
+
             this.scheduleReconnect();
         });
-        ws.on('error', (err) => {
-            console.warn('[Pilot] Tunnel error:', err.message);
-            // 'close' will follow; reconnect is scheduled there.
-        });
     }
 
     private cleanupAfterDisconnect(): void {
@@ -612,3 +645,23 @@ export function persistToken(token: string): void {
     }
 }
 
+/**
+ * Remove the persisted long-lived tunnel token from disk. Used when the
+ * control instance rejects the stored credential at upgrade so the agent
+ * can fall back to SENCHO_ENROLL_TOKEN without re-poisoning on restart.
+ * ENOENT is normal when no file was written yet.
+ *
+ * Exposed for unit tests.
+ */
+export function clearPersistedToken(): void {
+    try {
+        fs.unlinkSync(TOKEN_PATH);
+    } catch (err) {
+        const code = (err as NodeJS.ErrnoException).code;
+        if (code === 'ENOENT') return;
+        console.warn(
+            `[Pilot] Failed to remove persisted tunnel token at ${sanitizeForLog(TOKEN_PATH)} (${sanitizeForLog(code ?? 'unknown')}: ${sanitizeForLog((err as Error).message)})`,
+        );
+    }
+}
+