diff --git a/README.md b/README.md index 01812c3..65b0e1e 100644 --- a/README.md +++ b/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 ` - Cancel a running task. Example: `altis-cli app tasks cancel --yes` * `tasks logs ` - Stream logs for a running task. Example: `altis-cli app tasks logs ` + * `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 --from ` - Import database from another app. Example: `altis-cli app import database example-dev-01 --from example-stag-01` diff --git a/lib/authorize.ts b/lib/authorize.ts index 2554652..92e97e3 100644 --- a/lib/authorize.ts +++ b/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,6 +45,7 @@ class Authenticator { client_secret: this.options.secret, }; const opts = { + signal: AbortSignal.timeout( 30_000 ), method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded', @@ -52,14 +53,19 @@ class Authenticator { 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; } } diff --git a/lib/commands/stack/database/cli.ts b/lib/commands/stack/database/cli.ts new file mode 100644 index 0000000..85ba1f3 --- /dev/null +++ b/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; diff --git a/lib/commands/stack/database/index.ts b/lib/commands/stack/database/index.ts new file mode 100644 index 0000000..b4b960c --- /dev/null +++ b/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; diff --git a/lib/commands/stack/database/sequel.ts b/lib/commands/stack/database/sequel.ts new file mode 100644 index 0000000..3cd1ef6 --- /dev/null +++ b/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; diff --git a/lib/commands/stack/database/tunnel.ts b/lib/commands/stack/database/tunnel.ts new file mode 100644 index 0000000..bd2bea1 --- /dev/null +++ b/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; diff --git a/lib/commands/stack/ssh.ts b/lib/commands/stack/ssh.ts index 4ad57b9..58674db 100644 --- a/lib/commands/stack/ssh.ts +++ b/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 { 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 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)[key] === 'string' && (data as Record)[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((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() }; +} diff --git a/lib/database.ts b/lib/database.ts new file mode 100644 index 0000000..7165350 --- /dev/null +++ b/lib/database.ts @@ -0,0 +1,165 @@ +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtemp, writeFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import open from 'open'; +import pkg from '@humanmade/ssm'; +import { forwardSession, parseSession } from './database-tunnel.ts'; +import type { PortSession } from './database-tunnel.ts'; +import type { CliArguments, CommandBuilder } from './types/cli.ts'; +import type { DatabaseServerResponse } from './types/rest-api.ts'; +import { fetchJSON, getStack } from './commands/stack/util.ts'; +import Vantage from './vantage.ts'; + +interface DatabaseSession { + id: number; + aws_ssm_session_id: string; + stream_url: string; + token: string; +} +type DatabaseCredentials = Pick; +type SessionFactory = (url: string, id: string, token: string) => PortSession; + +function sessionFactory(): SessionFactory { + const Session = pkg.default; + const prototype = Session.prototype as unknown as Partial; + if (typeof prototype.writeRaw !== 'function' || typeof prototype.sendHandshakeResponse !== 'function') { + throw new Error('Database tunnels require the socket support in @humanmade/ssm 0.0.4 or newer; the installed version does not support it.'); + } + return (url, id, token) => new Session(url, id, token) as unknown as PortSession; +} + +export function databaseOptions(parser: CommandBuilder) { + return parser.option('port', { describe: 'Local loopback port to listen on.', type: 'number', default: 3306 }); +} + +/** Create one tunnel and terminate the owned AWS session on every exit path. */ +export async function createDatabaseTunnel(api: Vantage, stack: string, port: number, createSession = sessionFactory()) { + const url = `stack/applications/${encodeURIComponent(stack)}/database/sessions`; + const record = await fetchJSON(api, url, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ local_port: port }), + }); + if (!record || !Number.isInteger(record.id) || record.id < 1) { + throw new Error('Invalid database session response: missing Vantage session ID.'); + } + const cleanup = () => fetchJSON(api, `${url}/${record.id}`, { method: 'DELETE' }); + let forwarding: ReturnType; + try { + const data = parseSession(JSON.stringify({ id: record.aws_ssm_session_id, stream_url: record.stream_url, token: record.token })); + forwarding = forwardSession(createSession(data.stream_url, data.id, data.token), port); + } catch (error) { + await cleanup(); + throw error; + } + const done = forwarding.done.finally(cleanup); + try { + await Promise.race([ + once(forwarding.server, 'listening'), + done.then(() => { throw new Error('Database tunnel closed before it started listening.'); }), + ]); + } catch (error) { + forwarding.close(); + await done.catch(() => {}); + throw error; + } + return { port, done, close: () => { forwarding.close(); return done; } }; +} + +function mysqlQuote(value: string) { + return '"' + value.replace(/\\/g, '\\\\').replace(/"/g, '\\"').replace(/\n/g, '\\n').replace(/\r/g, '\\r').replace(/\t/g, '\\t') + '"'; +} + +export function mysqlConfig(credentials: DatabaseCredentials, port: number) { + const options = { host: '127.0.0.1', port: String(port), protocol: 'TCP', + user: credentials.username, password: credentials.password, database: credentials.database }; + return '[client]\n' + Object.entries(options).map(([key, value]) => `${key}=${mysqlQuote(value)}`).join('\n') + '\n'; +} + +function xml(value: string) { + return value.replace(/&/g, '&').replace(//g, '>').replace(/"/g, '"').replace(/'/g, '''); +} + +export function sequelConfig(credentials: DatabaseCredentials, port: number, stack: string) { + const fields = { name: stack, host: '127.0.0.1', user: credentials.username, + password: credentials.password, database: credentials.database, rdbms_type: 'mysql', type: 'StandardConnection' }; + return ` + + +auto_connect +ContentFilters +dataconnection +${Object.entries(fields).map(([key, value]) => `${key}${xml(value)}`).join('\n')} +port${port} +session +queryFavoritesqueryHistory +rdbms_typemysql +encryptedformatconnection +version1 +\n`; +} + +/** Keep credentials out of process arguments, environment variables, and logs. */ +export async function withConnectionFile(filename: string, content: string, run: (file: string) => Promise) { + const directory = await mkdtemp(path.join(tmpdir(), 'altis-database-')); + try { + const file = path.join(directory, filename); + await writeFile(file, content, { mode: 0o600 }); + return await run(file); + } finally { + await rm(directory, { recursive: true, force: true }); + } +} + +export async function launchMysql(file: string, tunnel: { port: number; done: Promise; close(): Promise }, launch = spawn) { + const child = launch('mysql', [`--defaults-file=${file}`, '--protocol=TCP', '--host=127.0.0.1', `--port=${tunnel.port}`], { stdio: 'inherit' }); + try { + const exited = new Promise((resolve, reject) => { + child.once('error', (error: NodeJS.ErrnoException) => reject(new Error(error.code === 'ENOENT' + ? 'The mysql client is not installed or is not on PATH.' : 'Could not start the mysql client.'))); + child.once('exit', (code, signal) => code === 0 || signal === 'SIGINT' ? resolve() + : reject(new Error(`mysql exited ${signal ? `with signal ${signal}` : `with status ${code}`}.`))); + }); + await Promise.race([exited, tunnel.done.then(() => exited)]); + } finally { + if (child.exitCode === null && child.signalCode === null) child.kill(); + await tunnel.close(); + } +} + +export async function runDatabase(argv: CliArguments, mode: 'tunnel' | 'cli' | 'sequel') { + const port = argv.port ?? 3306; + if (!Number.isInteger(port) || port < 1 || port > 65535) { + throw new Error('--port must be an integer between 1 and 65535.'); + } + if (mode === 'sequel' && process.platform !== 'darwin') { + throw new Error('database sequel is supported on macOS. Use database tunnel with your preferred database client.'); + } + const createSession = sessionFactory(); + const stack = await getStack(argv); + const api = new Vantage(argv.config); + let credentials: DatabaseCredentials | undefined; + if (mode !== 'tunnel') { + credentials = await fetchJSON(api, `stack/applications/${encodeURIComponent(stack)}/database-server`); + if (!credentials || typeof credentials.username !== 'string' || !credentials.username + || typeof credentials.database !== 'string' || typeof credentials.password !== 'string' || !credentials.password) { + throw new Error('Database credentials are not available to this account. Use database tunnel with your own credentials.'); + } + } + const tunnel = await createDatabaseTunnel(api, stack, port, createSession); + console.log(`Database tunnel for ${stack} listening on 127.0.0.1:${port}. Ctrl-C to close.`); + try { + if (mode === 'cli') { + await withConnectionFile('client.cnf', mysqlConfig(credentials!, port), file => launchMysql(file, tunnel)); + } else if (mode === 'sequel') { + await withConnectionFile('connection.spf', sequelConfig(credentials!, port, stack), async file => { + await open(file, { wait: false }); + await tunnel.done; + }); + } else { + await tunnel.done; + } + } finally { + await tunnel.close(); + } +} diff --git a/lib/types/cli.ts b/lib/types/cli.ts index 9942366..1ee6c4a 100644 --- a/lib/types/cli.ts +++ b/lib/types/cli.ts @@ -47,6 +47,7 @@ export interface CliArguments { origin?: string; output?: string; path?: string; + port?: number; pattern?: (string | number)[]; postSync?: boolean; replace?: (string | number)[]; diff --git a/lib/vantage.ts b/lib/vantage.ts index aa0044c..9c6258b 100644 --- a/lib/vantage.ts +++ b/lib/vantage.ts @@ -31,11 +31,11 @@ const REGIONS = [ ]; const STACK_CACHE_LIFETIME = 24 * 60 * 60 * 1000; -const AUTH_ENDPOINT = 'https://dashboard.altis-dxp.com/wp-login.php?action=oauth2_authorize'; const BASE_URL = process.env.VANTAGE_URL || 'https://dashboard.altis-dxp.com/api/'; +const AUTH_ENDPOINT = new URL('/wp-login.php?action=oauth2_authorize', BASE_URL).href; const TOKEN_ENDPOINT = `${ BASE_URL }oauth2/access_token`; -const OAUTH_KEY = 'za6qrd2teuev'; -const OAUTH_SECRET = 'kZCn1dluko2lout6gnDqsGqhYn9a7ICFX8YdOhBSd8hhD1w3'; +const OAUTH_KEY = process.env.VANTAGE_CLIENT_ID || 'za6qrd2teuev'; +const OAUTH_SECRET = process.env.VANTAGE_CLIENT_SECRET || 'kZCn1dluko2lout6gnDqsGqhYn9a7ICFX8YdOhBSd8hhD1w3'; const OAUTH_PORT = 7277; // ASCII "H" + "M" class VantageAPI { @@ -109,9 +109,9 @@ class VantageAPI { }, }; this.authPromise = authorize( opts ) - .then( token => { + .then( async token => { // Store for later use. - this.config.set( 'vantage_token', token ); + await this.config.set( 'vantage_token', token ); return { token }; } ); return this.authPromise; diff --git a/package-lock.json b/package-lock.json index 3c90b19..bc667b2 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,7 +10,7 @@ "license": "MIT", "dependencies": { "@automattic/vip-search-replace": "^2.0.0", - "@humanmade/ssm": "^0.0.3", + "@humanmade/ssm": "^0.0.4", "ansi-escapes": "^2.0.0", "application-config": "^1.0.1", "base64-stream": "^1.0.0", @@ -78,14 +78,14 @@ } }, "node_modules/@humanmade/ssm": { - "version": "0.0.3", - "resolved": "https://registry.npmjs.org/@humanmade/ssm/-/ssm-0.0.3.tgz", - "integrity": "sha512-3APnrGDZ/NnrxJSJIEnyKmQ4iq3xHxxwb5OoCKQAqNrrtTaJj/LPoxQP6a2EDAI48EO2czIIgiKIGGRMcsUn9w==", + "version": "0.0.4", + "resolved": "https://registry.npmjs.org/@humanmade/ssm/-/ssm-0.0.4.tgz", + "integrity": "sha512-7IlBz+NcAYNWyTNwCson3CzaOGLKj+uM+oURCZOAD3Ne/G82K6M5ukEZY78GXlWZhkq0v2HsMrm2iJjekhAnbg==", "license": "Apache-2.0", "dependencies": { "isomorphic-ws": "^4.0.1", "js-sha256": "^0.9.0", - "uuid": "^8.3.2", + "uuid": "^11.1.1", "ws": "^7.4.5" } }, diff --git a/package.json b/package.json index f512915..5031b32 100644 --- a/package.json +++ b/package.json @@ -35,7 +35,7 @@ }, "dependencies": { "@automattic/vip-search-replace": "^2.0.0", - "@humanmade/ssm": "^0.0.3", + "@humanmade/ssm": "^0.0.4", "ansi-escapes": "^2.0.0", "application-config": "^1.0.1", "base64-stream": "^1.0.0", diff --git a/scripts/live-ssm.js b/scripts/live-ssm.js index d50b35d..13577f1 100644 --- a/scripts/live-ssm.js +++ b/scripts/live-ssm.js @@ -94,9 +94,14 @@ export class LiveSession { async close() { if (this.disconnected) return; - this.session.close(); - try { await this.waitClosed(); } - finally { if (!this.disconnected) this.session.webSocket?.terminate?.(); } + try { + // Closing the WebSocket can leave docker exec's shell running. + this.session.write('exit\n'); + await this.waitClosed(); + } finally { + this.session.close(); + if (!this.disconnected) this.session.webSocket?.terminate?.(); + } } } diff --git a/test-native/cli.test.js b/test-native/cli.test.js index f0ffdad..3af1c4e 100644 --- a/test-native/cli.test.js +++ b/test-native/cli.test.js @@ -37,6 +37,9 @@ test('native source executable reads package metadata without a build', () => { test('native command discovery imports existing commands and nested namespaces', () => { assert.match(run(['--help']), /Commands:/); for (const scope of ['app', 'stack']) { + for (const action of ['tunnel', 'cli', 'sequel']) { + assert.match(run([scope, 'database', action, '--help']), new RegExp(`database ${action}`)); + } assert.match(run([scope, 'variables', '--help']), /variables \[stack\]/); assert.match(run([scope, 'sync-local', 'database', 'example-dev', '--help']), /database/); assert.match(run([scope, 'xray', 'list-traces', '--help']), /list-traces/); diff --git a/test/authorize.test.js b/test/authorize.test.js new file mode 100644 index 0000000..c9b84a6 --- /dev/null +++ b/test/authorize.test.js @@ -0,0 +1,97 @@ +import assert from 'node:assert/strict'; +import http from 'node:http'; +import { once } from 'node:events'; +import test from 'node:test'; +import { Authenticator } from '../dist/lib/authorize.js'; + +const options = exchange => ({ + endpoints: { authorize: 'https://staging.example/wp-login.php?action=oauth2_authorize', exchange }, + key: 'test-client', secret: 'test-secret', port: 0, + render: (title, message = '') => `${title} ${message}`, +}); +async function endpoint(t, handler) { + const server = http.createServer(handler); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + t.after(() => { server.closeAllConnections(); server.close(); }); + return `http://127.0.0.1:${server.address().port}/api/oauth2/access_token`; +} + +test('OAuth login and exchange follow VANTAGE_URL and authentication waits for token persistence', async t => { + const previous = { VANTAGE_URL: process.env.VANTAGE_URL, VANTAGE_CLIENT_ID: process.env.VANTAGE_CLIENT_ID, VANTAGE_CLIENT_SECRET: process.env.VANTAGE_CLIENT_SECRET }; + process.env.VANTAGE_URL = 'https://staging.example/api/'; + process.env.VANTAGE_CLIENT_ID = 'staging-test-client'; + process.env.VANTAGE_CLIENT_SECRET = 'staging-test-secret'; + t.after(() => { for (const [key, value] of Object.entries(previous)) { if (value === undefined) delete process.env[key]; else process.env[key] = value; } }); + let endpoints; + t.mock.method(Authenticator.prototype, 'start', function() { + endpoints = this.options.endpoints; + assert.equal(this.options.key, 'staging-test-client'); + assert.equal(this.options.secret, 'staging-test-secret'); + return Promise.resolve('synthetic-token'); + }); + const { default: Vantage } = await import('../dist/lib/vantage.js'); + let save; + let completed = false; + const saving = new Promise(resolve => { save = resolve; }); + const client = new Vantage({ get() {}, set(key, value) { + assert.equal(key, 'vantage_token'); assert.equal(value, 'synthetic-token'); return saving; + } }); + const authenticating = client.authenticate().then(value => { completed = true; return value; }); + await new Promise(resolve => setImmediate(resolve)); + assert.deepEqual(endpoints, { + authorize: 'https://staging.example/wp-login.php?action=oauth2_authorize', + exchange: 'https://staging.example/api/oauth2/access_token', + }); + assert.equal(completed, false); + save(); + assert.deepEqual(await authenticating, { token: 'synthetic-token' }); +}); + +test('token exchange rejects HTTP errors, invalid JSON, and missing tokens without echoing responses', async t => { + for (const [status, body, expected] of [ + [401, 'private server response', /HTTP 401/], + [200, 'private server response', /invalid JSON/], + [200, '{"error":"private server response"}', /did not return an access token/], + [200, '{"access_token":""}', /did not return an access token/], + ]) { + const url = await endpoint(t, (_req, res) => { res.writeHead(status); res.end(body); }); + const auth = new Authenticator(options(url)); + await assert.rejects(auth.exchangeCode('test-code'), error => expected.test(error.message) && !error.message.includes('private server response')); + } +}); + +test('callback only displays success after receiving a valid token', async t => { + let answer; + const requested = new Promise(resolve => { answer = resolve; }); + let response; + const url = await endpoint(t, (_req, res) => { response = res; answer(); }); + const auth = new Authenticator(options(url)); + let status, page; + const result = auth.handleRequest({ url: `/?code=test-code&state=${encodeURIComponent(auth.state)}` }, { + setHeader() {}, writeHead(value) { status = value; }, end(value) { page = value; }, + }); + await requested; + assert.equal(page, undefined); + response.end('{"access_token":"synthetic-token"}'); + assert.equal(await result, 'synthetic-token'); + assert.equal(status, 200); + assert.match(page, /Authenticated altis-cli!/); +}); + +test('failed callback rejects the login promise, displays failure, and closes the callback server', async t => { + const url = await endpoint(t, (_req, res) => { res.writeHead(401); res.end('{}'); }); + const auth = new Authenticator(options(url)); + let browser; + t.mock.method(auth, 'startAuthorization', async () => { + browser = fetch(`http://127.0.0.1:${auth.server.address().port}/?code=test-code&state=${encodeURIComponent(auth.state)}`); + await browser; + }); + await assert.rejects(auth.start(), /HTTP 401/); + const response = await browser; + assert.equal(response.status, 502); + const page = await response.text(); + assert.match(page, /Authentication failed/); + assert.doesNotMatch(page, /Authenticated altis-cli!/); + assert.equal(auth.server.listening, false); +}); diff --git a/test/commands.test.js b/test/commands.test.js index abd3b30..3ea1674 100644 --- a/test/commands.test.js +++ b/test/commands.test.js @@ -14,7 +14,7 @@ const commands = [ ['import', 'database', 'example-dev'], ['sync-local', 'all', 'example-dev'], ['sync-local', 'database', 'example-dev'], ['sync-local', 'uploads', 'example-dev'], ['xray', 'list-traces'], ['xray', 'trace', 'trace-id'], ['xray', 'trace-file', 'trace.json'], - ['ssh'], ['scp', 'app:/tmp/file', './file'], + ['database', 'tunnel'], ['database', 'cli'], ['database', 'sequel'], ['ssh'], ['scp', 'app:/tmp/file', './file'], ]; test('application resolver retains the existing stack argument', async () => { diff --git a/test/database-tunnel.test.js b/test/database-tunnel.test.js new file mode 100644 index 0000000..9500870 --- /dev/null +++ b/test/database-tunnel.test.js @@ -0,0 +1,165 @@ +import assert from 'node:assert/strict'; +import { EventEmitter, once } from 'node:events'; +import net from 'node:net'; +import test from 'node:test'; +import pkg from '@humanmade/ssm'; +import { forwardSession, parseSession } from '../dist/lib/database-tunnel.js'; +import command from '../dist/lib/commands/stack/database/tunnel.js'; + +class Session extends EventEmitter { + writes = []; + closed = 0; + responses = []; + writeRaw(data) { this.writes.push(Buffer.from(data)); this.emit('written'); } + sendHandshakeResponse(data) { this.responses.push(data); } + close() { this.closed++; this.emit('disconnect'); } +} +const handshake = { RequestedClientActions: [{ ActionType: 'SessionType', ActionParameters: { + SessionType: 'Port', Properties: { type: 'LocalPortForwarding' }, +} }] }; +async function start(t, port = 0) { + const session = new Session(); + const forwarding = forwardSession(session, port); + t.after(() => forwarding.close()); + session.emit('handshake', handshake); + const listening = once(forwarding.server, 'listening'); + session.emit('handshakeComplete', {}); + await listening; + return { session, ...forwarding }; +} +async function connect(t, server) { + const socket = net.connect(server.address().port, '127.0.0.1'); + t.after(() => socket.destroy()); + await once(socket, 'connect'); + return socket; +} + +test('validates session input without echoing credentials in errors', () => { + const valid = { id: 'session', token: 'secret', stream_url: 'wss://example.com/session' }; + assert.deepEqual(parseSession(JSON.stringify(valid)), valid); + for (const input of ['secret', 'null', '{}', JSON.stringify({ ...valid, token: '' }), + JSON.stringify({ ...valid, stream_url: 'https://example.com' })]) { + assert.throws(() => parseSession(input), error => !error.message.includes('secret')); + } +}); + +test('installed SSM provides the binary socket and handshake APIs', () => { + assert.equal(typeof pkg.default.prototype.writeRaw, 'function'); + assert.equal(typeof pkg.default.prototype.sendHandshakeResponse, 'function'); +}); + +test('an older SSM reports missing port support before requesting a session or opening a socket', async () => { + const prototype = pkg.default.prototype; + const original = Object.getOwnPropertyDescriptor(prototype, 'writeRaw'); + try { + delete prototype.writeRaw; + await assert.rejects(command.handler({ config: {}, port: 3306 }), /socket support/); + } finally { + Object.defineProperty(prototype, 'writeRaw', original); + } +}); + +test('validates local port before requesting a session', async () => { + for (const port of [0, -1, 65536, 1.5, NaN]) { + await assert.rejects(command.handler({ config: {}, port }), /--port/); + } +}); + +test('round-trips binary payloads larger than the old 8KB buffer and buffers the greeting', async t => { + const { session, server, done } = await start(t); + const greeting = Buffer.from([0, 255, 128, 1, 0]); + session.emit('rawOutput', greeting); + const socket = await connect(t, server); + const received = []; + let bytes = 0; + const payload = Buffer.alloc(128 * 1024); + for (let i = 0; i < payload.length; i++) payload[i] = i % 256; + const all = new Promise(resolve => socket.on('data', data => { + received.push(data); bytes += data.length; + if (bytes === greeting.length + payload.length) resolve(); + })); + session.emit('rawOutput', payload); + await all; + assert.deepEqual(Buffer.concat(received), Buffer.concat([greeting, payload])); + const written = once(session, 'written'); + socket.write(payload); + await written; + // TCP may split the input; wait until all chunks have reached SSM. + while (Buffer.concat(session.writes).length < payload.length) await once(session, 'written'); + assert.deepEqual(Buffer.concat(session.writes), payload); + socket.destroy(); + await done; + assert.equal(session.closed, 1); +}); + +test('rejects a second client without sharing the database stream', async t => { + const { session, server, done } = await start(t); + const first = await connect(t, server); + const second = await connect(t, server); + await once(second, 'close'); + assert.equal(first.destroyed, false); + assert.equal(session.closed, 0); + first.destroy(); + await done; +}); + +test('pauses local writes until SSM resumes', async t => { + const { session, server, done } = await start(t); + const socket = await connect(t, server); + session.emit('pause'); + socket.write('query'); + await new Promise(resolve => setTimeout(resolve, 20)); + assert.equal(session.writes.length, 0); + const written = once(session, 'written'); + session.emit('resume'); + await written; + assert.equal(Buffer.concat(session.writes).toString(), 'query'); + socket.destroy(); + await done; +}); + +test('bind failure rejects and closes the SSM session', async t => { + const occupied = net.createServer(); + occupied.listen(0, '127.0.0.1'); + await once(occupied, 'listening'); + t.after(() => occupied.close()); + const session = new Session(); + const forwarding = forwardSession(session, occupied.address().port); + const rejected = assert.rejects(forwarding.done, { code: 'EADDRINUSE' }); + session.emit('handshake', handshake); + session.emit('handshakeComplete', {}); + await rejected; + assert.equal(session.closed, 1); +}); + +test('unsupported actions fail handshake and release process listeners', async () => { + const count = process.listenerCount('SIGINT'); + const session = new Session(); + const { done } = forwardSession(session, 0); + const rejected = assert.rejects(done, /Unsupported/); + session.emit('handshake', { RequestedClientActions: [...handshake.RequestedClientActions, { ActionType: 'KMSEncryption' }] }); + await rejected; + assert.equal(session.responses[0].ProcessedClientActions[1].ActionStatus, 3); + assert.equal(session.closed, 1); + assert.equal(process.listenerCount('SIGINT'), count); +}); + +test('stops if output exceeds buffer limit before a client connects', async () => { + const session = new Session(); + const { done } = forwardSession(session, 0); + const rejected = assert.rejects(done, /buffer limit/); + session.emit('rawOutput', Buffer.alloc(16 * 1024 * 1024 + 1)); + await rejected; + assert.equal(session.closed, 1); +}); + +test('remote disconnect closes the local listener and client', async t => { + const { session, server, done } = await start(t); + const socket = await connect(t, server); + const closed = once(socket, 'close'); + const rejected = assert.rejects(done, /lost connection/); + session.emit('disconnect', 'lost connection'); + await rejected; + await closed; + assert.equal(server.listening, false); +}); diff --git a/test/database.test.js b/test/database.test.js new file mode 100644 index 0000000..56c338c --- /dev/null +++ b/test/database.test.js @@ -0,0 +1,158 @@ +import assert from 'node:assert/strict'; +import { EventEmitter, once } from 'node:events'; +import { access, readFile, stat } from 'node:fs/promises'; +import net from 'node:net'; +import path from 'node:path'; +import test from 'node:test'; +import { Response } from 'node-fetch'; +import { createDatabaseTunnel, launchMysql, mysqlConfig, sequelConfig, withConnectionFile } from '../dist/lib/database.js'; + +const handshake = { RequestedClientActions: [{ ActionType: 'SessionType', ActionParameters: { + SessionType: 'Port', Properties: { type: 'LocalPortForwarding' }, +} }] }; +class Session extends EventEmitter { + closed = 0; + writeRaw() {} + sendHandshakeResponse() {} + close() { this.closed++; this.emit('disconnect'); } +} +function apiFixture(record = { id: 123, aws_ssm_session_id: 'aws-session', stream_url: 'wss://example.test', token: 'secret' }) { + const calls = []; + return { + calls, + fetch: async (url, options) => { + calls.push({ url, options }); + return options.method === 'DELETE' ? new Response(null, { status: 204 }) : new Response(JSON.stringify(record)); + }, + }; +} +function sessionFactory() { + const session = new Session(); + queueMicrotask(() => { + session.emit('handshake', handshake); + session.emit('handshakeComplete'); + }); + return session; +} +async function unusedPort() { + const server = net.createServer(); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const port = server.address().port; + await new Promise(resolve => server.close(resolve)); + return port; +} + +test('Vantage tunnel uses the AWS ID for transport and Vantage ID for cleanup exactly once', async () => { + const api = apiFixture(); + const port = await unusedPort(); + const tunnel = await createDatabaseTunnel(api, 'example-app', port, (url, id, token) => { + assert.equal(url, 'wss://example.test'); + assert.equal(id, 'aws-session'); + assert.equal(token, 'secret'); + return sessionFactory(); + }); + assert.deepEqual(JSON.parse(api.calls[0].options.body), { local_port: port }); + assert.equal(api.calls[0].url, 'stack/applications/example-app/database/sessions'); + await tunnel.close(); + await tunnel.close(); + assert.equal(api.calls.length, 2); + assert.equal(api.calls[1].options.method, 'DELETE'); + assert.equal(api.calls[1].url, 'stack/applications/example-app/database/sessions/123'); +}); + +test('invalid connection data triggers remote cleanup without exposing the token', async () => { + const api = apiFixture({ id: 123, aws_ssm_session_id: 'aws-session', stream_url: 'https://invalid.test', token: 'secret' }); + await assert.rejects(createDatabaseTunnel(api, 'example-app', 3306, sessionFactory), error => !error.message.includes('secret')); + assert.equal(api.calls.at(-1).options.method, 'DELETE'); +}); + +test('a session constructor failure triggers remote cleanup', async () => { + const api = apiFixture(); + await assert.rejects(createDatabaseTunnel(api, 'example-app', 3306, () => { throw new Error('failed'); }), /failed/); + assert.equal(api.calls.at(-1).options.method, 'DELETE'); +}); + +test('bind failure terminates the remote session', async t => { + const occupied = net.createServer(); + occupied.listen(0, '127.0.0.1'); + await once(occupied, 'listening'); + t.after(() => occupied.close()); + const api = apiFixture(); + await assert.rejects(createDatabaseTunnel(api, 'example-app', occupied.address().port, sessionFactory), { code: 'EADDRINUSE' }); + assert.equal(api.calls.filter(call => call.options.method === 'DELETE').length, 1); +}); + +test('disconnect before handshake fails promptly and terminates the remote session', async () => { + const api = apiFixture(); + await assert.rejects(createDatabaseTunnel(api, 'example-app', 3306, () => { + const session = new Session(); + queueMicrotask(() => session.emit('disconnect', 'Connection lost')); + return session; + }), /Connection lost/); + assert.equal(api.calls.at(-1).options.method, 'DELETE'); +}); + +test('connection files are private and removed after success and failure', async () => { + for (const fail of [false, true]) { + let filename; + const operation = withConnectionFile('client.cnf', 'password=secret', async file => { + filename = file; + assert.equal((await stat(file)).mode & 0o777, 0o600); + assert.equal((await stat(path.dirname(file))).mode & 0o777, 0o700); + assert.equal(await readFile(file, 'utf8'), 'password=secret'); + if (fail) throw new Error('launch failed'); + }); + if (fail) await assert.rejects(operation, /launch failed/); + else await operation; + await assert.rejects(access(filename)); + await assert.rejects(access(path.dirname(filename))); + } +}); + +test('mysql options quote credentials and force TCP to the loopback tunnel', () => { + const config = mysqlConfig({ username: 'admin', password: 'a"\\b\n!include /tmp/other', database: 'wordpress' }, 3307); + assert.match(config, /host="127\.0\.0\.1"/); + assert.match(config, /port="3307"/); + assert.match(config, /protocol="TCP"/); + assert.ok(config.includes('password="a\\"\\\\b\\n!include /tmp/other"')); + assert.ok(!config.includes('\n!include')); +}); + +test('Sequel connection uses loopback with XML-escaped credentials and no SSH settings', () => { + const config = sequelConfig({ username: 'a&b', password: '"', database: 'wordpress' }, 3307, 'example-prod'); + assert.ok(config.includes('host127.0.0.1')); + assert.ok(config.includes('port3307')); + assert.ok(config.includes('a&b')); + assert.ok(config.includes('<secret>"')); + assert.ok(!config.includes('ssh_host')); +}); + +test('mysql launch inherits the terminal and never passes credentials in argv or environment', async () => { + const calls = []; + let closed = 0; + await launchMysql('/tmp/private/client.cnf', { port: 3307, done: new Promise(() => {}), close: async () => { closed++; } }, (...args) => { + calls.push(args); + const child = new EventEmitter(); + child.exitCode = null; child.signalCode = null; + child.kill = () => assert.fail('a finished client should not be killed'); + queueMicrotask(() => { child.exitCode = 0; child.emit('exit', 0, null); }); + return child; + }); + assert.deepEqual(calls, [['mysql', ['--defaults-file=/tmp/private/client.cnf', '--protocol=TCP', '--host=127.0.0.1', '--port=3307'], { stdio: 'inherit' }]]); + assert.equal(closed, 1); +}); + +test('missing mysql client reports an actionable error and closes the tunnel', async () => { + let closed = 0; + await assert.rejects(launchMysql('/tmp/private/client.cnf', { + port: 3307, done: new Promise(() => {}), close: async () => { closed++; }, + }, () => { + const child = new EventEmitter(); + child.exitCode = null; child.signalCode = null; + child.kill = () => {}; + queueMicrotask(() => child.emit('error', Object.assign(new Error(), { code: 'ENOENT' }))); + return child; + }), /mysql client is not installed/); + assert.equal(closed, 1); +}); diff --git a/test/live-ssm.test.js b/test/live-ssm.test.js new file mode 100644 index 0000000..6ef262a --- /dev/null +++ b/test/live-ssm.test.js @@ -0,0 +1,32 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import test from 'node:test'; +import { LiveSession } from '../scripts/live-ssm.js'; + +test('live runner waits for remote shell exit before closing its socket', async () => { + const session = new EventEmitter(); + const events = []; + session.write = command => { + events.push(command); + queueMicrotask(() => { + events.push('remote exit'); + session.emit('disconnect'); + }); + }; + session.close = () => events.push('socket close'); + const client = new LiveSession(session, 1000); + await client.close(); + await client.close(); + assert.deepEqual(events, ['exit\n', 'remote exit', 'socket close']); +}); + +test('live runner releases the socket when it cannot send remote exit', async () => { + const session = new EventEmitter(); + const events = []; + session.write = () => { throw new Error('socket unavailable'); }; + session.close = () => events.push('close'); + session.webSocket = { terminate: () => events.push('terminate') }; + const client = new LiveSession(session, 1000); + await assert.rejects(client.close(), /socket unavailable/); + assert.deepEqual(events, ['close', 'terminate']); +}); diff --git a/test/ssh.test.js b/test/ssh.test.js index cebd17b..3ee82e0 100644 --- a/test/ssh.test.js +++ b/test/ssh.test.js @@ -45,6 +45,17 @@ test('SSH escape sequence closes the active session', async () => { assert.equal(closed, true); }); +test('SSH input preserves UTF-8 characters split across terminal chunks', async () => { + const input = new PassThrough(); + const writes = []; + const detach = attachSessionInput({ write: value => writes.push(value), close() {} }, input, new PassThrough()); + const text = 'café 日本語 🐙\r'; + for (const byte of Buffer.from(text)) input.write(Buffer.from([byte])); + await new Promise(resolve => setImmediate(resolve)); + detach(); + assert.equal(writes.join(''), text); +}); + test('SSH keepalive pings the active session', () => { let callback; let pings = 0;