Add database tunnel, CLI, and Sequel commands via SSM

README.md

@@ -41,6 +41,9 @@ Always use `altis-cli help` for the most up-to-date list of commands.
 	* `tasks` - List, cancel, or stream running tasks. Example: `altis-cli app tasks example-dev-01`
 	* `tasks cancel <id>` - Cancel a running task. Example: `altis-cli app tasks cancel <id> --yes`
 	* `tasks logs <id>` - Stream logs for a running task. Example: `altis-cli app tasks logs <id>`
+	* `database tunnel [stack]` - Forward a local port to the primary database.
+	* `database cli [stack]` - Open the local `mysql` client through a tunnel.
+	* `database sequel [stack]` - Open Sequel Ace/Sequel Pro through a tunnel (macOS).
 	* `database-tables [stack]` - List database tables. Example: `altis-cli app database-tables example-dev-01`
 	* `uploads-prefixes [stack]` - List uploads prefixes. Example: `altis-cli app uploads-prefixes example-dev-01`
 	* `import database <target> --from <source>` - Import database from another app. Example: `altis-cli app import database example-dev-01 --from example-stag-01`

lib/authorize.ts

@@ -12,7 +12,7 @@ interface AuthorizationOptions {
 	port: number;
 	render: (title: string, message?: string) => string;
 }
-class Authenticator {
+export class Authenticator {
 	options: AuthorizationOptions;
 	server: http.Server;
 	state: string;
@@ -33,7 +33,7 @@ class Authenticator {
 		};
 
 		const url = `${ this.options.endpoints.authorize }&${ qs.stringify( params ) }`;
-		open( url );
+		return open( url );
 	}
 
 	exchangeCode( code: string ) {
@@ -45,21 +45,27 @@ class Authenticator {
 			client_secret: this.options.secret,
 		};
 		const opts = {
+			signal: AbortSignal.timeout( 30_000 ),
 			method: 'POST',
 			headers: {
 				'Content-Type': 'application/x-www-form-urlencoded',
 			},
 			body: qs.stringify( params ),
 		};
 		return fetch( this.options.endpoints.exchange, opts )
-			.then( resp => resp.json() )
+			.then( async resp => {
+				if ( ! resp.ok ) throw new Error( `OAuth token exchange failed (HTTP ${ resp.status }).` );
+				try { return await resp.json(); }
+				catch { throw new Error( 'OAuth token exchange returned invalid JSON.' ); }
+			} )
 			.then( data => {
-				const token = (data as { access_token: string }).access_token;
+				const token = data && typeof data === 'object' && 'access_token' in data ? data.access_token : undefined;
+				if ( typeof token !== 'string' || ! token.trim() ) throw new Error( 'OAuth token exchange did not return an access token.' );
 				return token;
 			} );
 	}
 
-	handleRequest( req: http.IncomingMessage, res: http.ServerResponse ) {
+	async handleRequest( req: http.IncomingMessage, res: http.ServerResponse ) {
 		const params = new URL( req.url || '/', `http://localhost:${ this.options.port }` ).searchParams;
 		const states = params.getAll( 'state' );
 		const code = params.getAll( 'code' ).join( ',' );
@@ -69,18 +75,24 @@ class Authenticator {
 		if ( states.length !== 1 || states[0] !== this.state ) {
 			res.writeHead( 400 );
 			res.end( this.options.render( 'Invalid state parameter.' ) );
-			return Promise.resolve( null );
+			return null;
 		}
 		if ( ! code ) {
 			res.writeHead( 400 );
 			res.end( this.options.render( 'Missing code parameter.' ) );
-			return Promise.resolve( null );
+			return null;
 		}
 
-		res.writeHead( 200 );
-		res.end( this.options.render( 'Authenticated altis-cli!', 'You can now close this window and go back to your terminal.' ) );
-
-		return this.exchangeCode( code );
+		try {
+			const token = await this.exchangeCode( code );
+			res.writeHead( 200 );
+			res.end( this.options.render( 'Authenticated altis-cli!', 'You can now close this window and go back to your terminal.' ) );
+			return token;
+		} catch ( error ) {
+			res.writeHead( 502 );
+			res.end( this.options.render( 'Authentication failed.', 'Token exchange failed. See your terminal for details.' ) );
+			throw error;
+		}
 	}
 
 	start() {
@@ -100,12 +112,12 @@ class Authenticator {
 					reject( new Error( `Port ${ this.options.port } is already in use.` ) );
 					return;
 				}
+				reject( err );
 			} );
 
-			this.server.listen( this.options.port, () => this.startAuthorization() );
+			this.server.listen( this.options.port, () => { this.startAuthorization().catch( reject ); } );
 		} );
-		this.promise.then( () => this.server.close() );
-		this.promise.catch( err => this.server.close() );
+		void this.promise.then( () => this.server.close(), () => this.server.close() );
 		return this.promise;
 	}
 }

lib/commands/stack/database/cli.ts

@@ -0,0 +1,9 @@
+import type { CommandDefinition } from '../../../types/cli.ts';
+import { databaseOptions, runDatabase } from '../../../database.ts';
+
+export default {
+	command: 'cli [stack]',
+	describe: 'Open the mysql command-line client through a database tunnel.',
+	builder: databaseOptions,
+	handler: argv => runDatabase(argv, 'cli'),
+} satisfies CommandDefinition;

lib/commands/stack/database/index.ts

@@ -0,0 +1,13 @@
+import { fileURLToPath } from 'node:url';
+import type { CommandBuilder, CommandDefinition } from '../../../types/cli.ts';
+import buildSubcommands from '../../../buildSubcommands.ts';
+
+export default {
+	command: 'database',
+	describe: 'Database commands.',
+	builder: async (command: CommandBuilder) => {
+		command.demandCommand(1);
+		const subcommands = await buildSubcommands(fileURLToPath(new URL('.', import.meta.url)));
+		command.command(subcommands);
+	},
+} satisfies CommandDefinition;

lib/commands/stack/database/sequel.ts

@@ -0,0 +1,9 @@
+import type { CommandDefinition } from '../../../types/cli.ts';
+import { databaseOptions, runDatabase } from '../../../database.ts';
+
+export default {
+	command: 'sequel [stack]',
+	describe: 'Open Sequel Ace or Sequel Pro through a database tunnel (macOS).',
+	builder: databaseOptions,
+	handler: argv => runDatabase(argv, 'sequel'),
+} satisfies CommandDefinition;

lib/commands/stack/database/tunnel.ts

@@ -0,0 +1,9 @@
+import type { CommandDefinition } from '../../../types/cli.ts';
+import { databaseOptions, runDatabase } from '../../../database.ts';
+
+export default {
+	command: 'tunnel [stack]',
+	describe: 'Forward a local port to the application’s primary database.',
+	builder: databaseOptions,
+	handler: argv => runDatabase(argv, 'tunnel'),
+} satisfies CommandDefinition;

lib/commands/stack/ssh.ts

@@ -5,6 +5,7 @@ import pkg from '@humanmade/ssm';
 const AWSSSMSession = pkg.default;
 import { getStack } from './util.ts';
 import Vantage from '../../vantage.ts';
+import { StringDecoder } from 'node:string_decoder';
 
 // Keep the socket alive by pinging every 30s.
 const IDLE_INTERVAL = 30_000;
@@ -15,10 +16,12 @@ export function attachSessionInput( session: Pick<InstanceType<typeof AWSSSMSess
 	let lastWasNewline = true;
 	let startedSshControl = false;
 	const wasRaw = Boolean( input.isRaw );
+	const decoder = new StringDecoder( 'utf8' );
 
 	const listener = (data: Buffer) => {
 		let msg = '';
-		for ( const byte of data.values() ) {
+		for ( const char of decoder.write( data ) ) {
+			const byte = char.codePointAt( 0 );
 			// ~ byte, indicates start of control code.
 			if ( lastWasNewline && byte === 0x7E && !startedSshControl ) {
 				startedSshControl = true;
@@ -27,8 +30,6 @@ export function attachSessionInput( session: Pick<InstanceType<typeof AWSSSMSess
 
 			// Check for newlines (LF or CR).
 			lastWasNewline = byte === 0x0A || byte === 0x0D;
-
-			const char = String.fromCharCode( byte );
 			if ( startedSshControl ) {
 				startedSshControl = false;
 				// ~.  - terminate connection (and any multiplexed sessions)

lib/database-tunnel.ts

@@ -0,0 +1,153 @@
+import { createServer, type Socket } from 'node:net';
+import { Duplex } from 'node:stream';
+
+interface SessionData {
+	id: string;
+	stream_url: string;
+	token: string;
+}
+interface HandshakeAction {
+	ActionType: string;
+	ActionParameters?: { SessionType?: string; Properties?: { type?: string } };
+}
+export interface PortSession {
+	on(event: string, listener: (...args: any[]) => void): unknown;
+	writeRaw(data: Uint8Array): void;
+	sendHandshakeResponse(response: unknown): void;
+	close(): void;
+}
+
+// Bound data waiting for a local client, or for a slow client to read.
+const MAX_BUFFER = 16 * 1024 * 1024;
+
+export function parseSession(raw: string): SessionData {
+	let data: unknown;
+	try {
+		data = JSON.parse(raw);
+	} catch {
+		throw new Error('Expected database session JSON (id, stream_url, token).');
+	}
+	if (!data || typeof data !== 'object' || !['id', 'stream_url', 'token'].every(
+		key => typeof (data as Record<string, unknown>)[key] === 'string' && (data as Record<string, string>)[key].length > 0
+	)) {
+		throw new Error('Session JSON must contain non-empty id, stream_url, and token strings.');
+	}
+	const session = data as SessionData;
+	try {
+		if (new URL(session.stream_url).protocol !== 'wss:') throw new Error();
+	} catch {
+		throw new Error('Session stream_url must be a valid wss:// URL.');
+	}
+	return session;
+}
+
+/** One non-multiplexed SSM session supports one database client connection. */
+export function forwardSession(session: PortSession, port = 3306) {
+	let stopped = false;
+	let accepted = false;
+	let negotiated = false;
+	let listeningStarted = false;
+	let paused = false;
+	let pendingWrite: (() => void) | undefined;
+	let client: Socket | undefined;
+	let resolveDone: () => void;
+	let rejectDone: (error: Error) => void;
+	const done = new Promise<void>((resolve, reject) => {
+		resolveDone = resolve;
+		rejectDone = reject;
+	});
+	const tunnel = new Duplex({
+		read() {},
+		write(chunk: Buffer, _encoding, callback) {
+			const write = () => {
+				try { session.writeRaw(chunk); callback(); }
+				catch (error) { callback(error as Error); }
+			};
+			if (paused) pendingWrite = write;
+			else write();
+		},
+	});
+	const server = createServer(socket => {
+		if (accepted || stopped) {
+			socket.destroy();
+			return;
+		}
+		accepted = true;
+		client = socket;
+		socket.on('error', stop);
+		socket.on('close', () => stop());
+		socket.pipe(tunnel).pipe(socket);
+	});
+	const stop = (error?: Error) => {
+		if (stopped) return;
+		stopped = true;
+		clearTimeout(timeout);
+		pendingWrite = undefined;
+		process.removeListener('SIGINT', interrupt);
+		process.removeListener('SIGTERM', interrupt);
+		client?.destroy();
+		tunnel.destroy();
+		server.close();
+		try { session.close(); }
+		catch (closeError) { error ??= closeError as Error; }
+		if (error) rejectDone(error);
+		else resolveDone();
+	};
+	const interrupt = () => stop();
+	const timeout = setTimeout(() => stop(new Error('Timed out negotiating the database session.')), 30_000);
+	process.on('SIGINT', interrupt);
+	process.on('SIGTERM', interrupt);
+	tunnel.on('error', stop);
+	server.on('error', stop);
+	session.on('disconnect', (reason?: string) => stop(new Error(reason || 'Database session disconnected.')));
+	session.on('pause', () => { paused = true; });
+	session.on('resume', () => {
+		paused = false;
+		const write = pendingWrite;
+		pendingWrite = undefined;
+		write?.();
+	});
+	session.on('rawOutput', (data: Uint8Array) => {
+		if (stopped) return;
+		if (tunnel.readableLength + data.byteLength > MAX_BUFFER) {
+			stop(new Error('Database client is too slow: tunnel buffer limit exceeded.'));
+			return;
+		}
+		tunnel.push(Buffer.from(data));
+	});
+	session.on('handshake', (data: { RequestedClientActions: HandshakeAction[] }) => {
+		if (stopped) return;
+		try {
+			let portSession = false;
+			const errors: string[] = [];
+			const actions = data.RequestedClientActions.map(action => {
+				const supported = action.ActionType === 'SessionType'
+					&& action.ActionParameters?.SessionType === 'Port'
+					&& action.ActionParameters.Properties?.type === 'LocalPortForwarding';
+				if (supported) portSession = true;
+				else errors.push(`Unsupported database session action: ${action.ActionType}`);
+				return { ActionType: action.ActionType, ActionStatus: supported ? 1 : 3,
+					ActionResult: null, Error: supported ? '' : errors[errors.length - 1] };
+			});
+			// Stay below 1.1.70: this transport does not implement SSM multiplexing.
+			session.sendHandshakeResponse({ ClientVersion: '1.1.69', ProcessedClientActions: actions, Errors: errors });
+			if (!portSession || errors.length) {
+				stop(new Error(errors.join('; ') || 'SSM did not request a port forwarding session.'));
+				return;
+			}
+			negotiated = true;
+		} catch (error) { stop(error as Error); }
+	});
+	session.on('handshakeComplete', () => {
+		if (stopped || listeningStarted) return;
+		if (!negotiated) {
+			stop(new Error('SSM completed the handshake without negotiating port forwarding.'));
+			return;
+		}
+		clearTimeout(timeout);
+		listeningStarted = true;
+		server.listen({ host: '127.0.0.1', port });
+	});
+	// Do not send application-level keepalive data into the database stream.
+	return { server, done, close: () => stop() };
+}