Skip to content

feat: Support Developer Edition connections #603

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
hessjcg wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat: Support Developer Edition connections #603

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
568 changes: 557 additions & 11 deletions package-lock.json
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,9 +87,12 @@
"url": "git+https://github.com/GoogleCloudPlatform/cloud-sql-nodejs-connector"
},
"dependencies": {
"@google-cloud/sql": "^0.25.0",
"@googleapis/sqladmin": "^37.0.0",
"gaxios": "^7.3.1",
"google-auth-library": "^10.9.1",
"@grpc/grpc-js": "^1.14.4",
"@grpc/proto-loader": "^0.8.1",
"p-throttle": "^8.1.0"
}
}
8 changes: 4 additions & 4 deletions scripts/fixup.cjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,12 +39,12 @@ async function fixupImportFileExtensions() {
const replaceExtension = source =>
source.replace(/(^import.*from '\.\/.*)(';)$/gm, '$1.js$2');

const recursiveReadDir = async path => {
const dirResults = await readdir(mjsDistFolder, {withFileTypes: true});
const recursiveReadDir = async dirPath => {
const dirResults = await readdir(dirPath, {withFileTypes: true});
for (const entry of dirResults) {
const path = resolve(mjsDistFolder, entry.name);
const path = resolve(dirPath, entry.name);
if (entry.isDirectory()) {
recursiveReadDir(path);
await recursiveReadDir(path);
} else if (path.endsWith('.js') || path.endsWith('.d.ts')) {
mjsFilePaths.push(path);
}
Expand Down
62 changes: 37 additions & 25 deletions src/cloud-sql-instance.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,10 +63,10 @@ interface CloudSQLInstanceOptions {
}

interface RefreshResult {
ephemeralCert: SslCert;
ephemeralCert?: SslCert;
host: string;
privateKey: string;
serverCaCert: SslCert;
privateKey?: string;
serverCaCert?: SslCert;
}

export class CloudSQLInstance {
Expand Down Expand Up @@ -243,10 +243,12 @@ export class CloudSQLInstance {
// then we go ahead and update values
this.updateValues(nextValues);

const refreshInterval = getRefreshInterval(
/* c8 ignore next */
String(this.ephemeralCert?.expirationTime)
);
let refreshInterval = 3600000; // 1 hour default
if (this.ephemeralCert) {
refreshInterval = getRefreshInterval(
this.ephemeralCert.expirationTime
);
}
this.scheduleRefresh(refreshInterval);

// This is the end of the successful refresh chain, so now
Expand Down Expand Up @@ -289,30 +291,37 @@ export class CloudSQLInstance {
return Promise.reject('closed');
}

const rsaKeys: RSAKeys = await generateKeys();
const metadata: InstanceMetadata =
await this.sqlAdminFetcher.getInstanceMetadata(this.instanceInfo);

const ephemeralCert = await this.sqlAdminFetcher.getEphemeralCertificate(
this.instanceInfo,
rsaKeys.publicKey,
this.authType
);
let host;
if (this.instanceInfo && this.instanceInfo.domainName) {
try {
const ips = await resolveARecord(this.instanceInfo.domainName);
if (ips && ips.length > 0) {
host = ips[0];
let host = '';
let ephemeralCert;
let privateKey;

if (this.ipType !== IpAddressTypes.SQL_DATA) {
const rsaKeys: RSAKeys = await generateKeys();
ephemeralCert = await this.sqlAdminFetcher.getEphemeralCertificate(
this.instanceInfo,
rsaKeys.publicKey,
this.authType
);
privateKey = rsaKeys.privateKey;

if (this.instanceInfo && this.instanceInfo.domainName) {
try {
const ips = await resolveARecord(this.instanceInfo.domainName);
if (ips && ips.length > 0) {
host = ips[0];
}
} catch (e) {
// ignore error, fallback to metadata IP
}
} catch (e) {
// ignore error, fallback to metadata IP
}
if (!host) {
host = selectIpAddress(metadata.ipAddresses, this.ipType);
}
}
if (!host) {
host = selectIpAddress(metadata.ipAddresses, this.ipType);
}
const privateKey = rsaKeys.privateKey;

const serverCaCert = metadata.serverCaCert;
this.serverCaMode = metadata.serverCaMode;
this.dnsName = metadata.dnsName;
Expand Down Expand Up @@ -347,6 +356,9 @@ export class CloudSQLInstance {
privateKey,
serverCaCert,
}: Partial<RefreshResult>): boolean {
if (this.ipType === IpAddressTypes.SQL_DATA) {
return true;
}
if (!ephemeralCert || !host || !privateKey || !serverCaCert) {
return false;
}
Expand Down
172 changes: 169 additions & 3 deletions src/connector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
// limitations under the License.

import {createServer, Server, Socket} from 'node:net';
import tls from 'node:tls';

import {promisify} from 'node:util';
import {AuthClient, GoogleAuth} from 'google-auth-library';
import {CloudSQLInstance} from './cloud-sql-instance';
Expand All @@ -22,6 +22,8 @@ import {IpAddressTypes} from './ip-addresses';
import {AuthTypes} from './auth-types';
import {SQLAdminFetcher} from './sqladmin-fetcher';
import {CloudSQLConnectorError} from './errors';
import {resolveInstanceName} from './parse-instance-connection-name';
import {SqlDataClient} from './sql-data-client';

// These Socket types are subsets from nodejs definitely typed repo, ref:
// https://github.com/DefinitelyTyped/DefinitelyTyped/blob/ae0fe42ff0e6e820e8ae324acf4f8e944aa1b2b7/types/node/v18/net.d.ts#L437
Expand All @@ -46,18 +48,20 @@ export declare interface ConnectionOptions {
domainName?: string;
failoverPeriod?: number;
limitRateInterval?: number;
sqlDataEndpoint?: string;
sqlDataStreamTimeout?: number;
}

export declare interface SocketConnectionOptions extends ConnectionOptions {
listenOptions: UnixSocketOptions;
}

interface StreamFunction {
(): tls.TLSSocket;
(): Socket;
}

interface PromisedStreamFunction {
(): Promise<tls.TLSSocket>;
(): Promise<Socket>;
}

// DriverOptions is the interface describing the object returned by
Expand Down Expand Up @@ -187,6 +191,8 @@ export interface ConnectorOptions {
*/
universeDomain?: string;
userAgent?: string;
sqlDataEndpoint?: string;
sqlDataStreamTimeout?: number;
}

// The Connector class is the main public API to interact
Expand All @@ -196,6 +202,10 @@ export class Connector {
private readonly sqlAdminFetcher: SQLAdminFetcher;
private readonly localProxies: Set<Server>;
private readonly sockets: Set<Socket>;
private readonly sqlDataEndpoint?: string;
private readonly sqlDataStreamTimeout?: number;
private readonly sqlDataTunnels = new Map<string, SqlDataClient>();
private readonly sqlDataUnsupportedInstances = new Set<string>();

constructor(opts: ConnectorOptions = {}) {
this.sqlAdminFetcher = new SQLAdminFetcher({
Expand All @@ -207,6 +217,8 @@ export class Connector {
this.instances = new CloudSQLInstanceMap(this.sqlAdminFetcher);
this.localProxies = new Set();
this.sockets = new Set();
this.sqlDataEndpoint = opts.sqlDataEndpoint;
this.sqlDataStreamTimeout = opts.sqlDataStreamTimeout;
}

// Connector.getOptions is a method that accepts a Cloud SQL instance
Expand All @@ -222,8 +234,162 @@ export class Connector {
// const res = await pool.query('SELECT * FROM pg_catalog.pg_tables;')
async getOptions(opts: ConnectionOptions): Promise<DriverOptions> {
const {instances} = this;

const instanceInfo = await resolveInstanceName(
opts.instanceConnectionName,
opts.domainName,
this.sqlAdminFetcher
);
const connectionName = `${instanceInfo.projectId}:${instanceInfo.regionId}:${instanceInfo.instanceId}`;

let ipType = opts.ipType || IpAddressTypes.PUBLIC;

if (
ipType === IpAddressTypes.SQL_DATA &&
this.sqlDataUnsupportedInstances.has(connectionName)
) {
ipType = IpAddressTypes.PUBLIC;
opts.ipType = ipType;
}

await instances.loadInstance(opts);

if (ipType === IpAddressTypes.SQL_DATA) {
let tunnel = this.sqlDataTunnels.get(connectionName);
if (!tunnel) {
const cloudSqlInstance = instances.getInstance(opts);
const getDirectSocket = () => {
const {
instanceInfo,
ephemeralCert,
host,
port,
privateKey,
serverCaCert,
dnsName,
} = cloudSqlInstance;

if (
instanceInfo &&
ephemeralCert &&
host &&
port &&
privateKey &&
serverCaCert
) {
const tlsSocket = getSocket({
instanceInfo,
ephemeralCert,
host,
port,
privateKey,
serverCaCert,
instanceDnsName: dnsName,
serverName: instanceInfo.domainName || dnsName,
});
tlsSocket.once('error', () => {
cloudSqlInstance.forceRefresh();
});
tlsSocket.once('secureConnect', async () => {
cloudSqlInstance.setEstablishedConnection();
});

cloudSqlInstance.addSocket(tlsSocket);

return tlsSocket;
}

throw new CloudSQLConnectorError({
message: 'Invalid Cloud SQL Instance info',
code: 'EBADINSTANCEINFO',
});
};

tunnel = new SqlDataClient({
instanceConnectionName: connectionName,
auth: this.sqlAdminFetcher.adminAuth,
endpoint: opts.sqlDataEndpoint || this.sqlDataEndpoint,
streamTimeout: opts.sqlDataStreamTimeout || this.sqlDataStreamTimeout,
getDirectSocket,
onUnsupported: () => {
this.sqlDataUnsupportedInstances.add(connectionName);
},
});
this.sqlDataTunnels.set(connectionName, tunnel);
}

const tunnelPort = await tunnel.start();

return {
stream: () => {
if (this.sqlDataUnsupportedInstances.has(connectionName)) {
const cloudSqlInstance = instances.getInstance(opts);
const {
instanceInfo,
ephemeralCert,
host,
port,
privateKey,
serverCaCert,
dnsName,
} = cloudSqlInstance;

if (
instanceInfo &&
ephemeralCert &&
host &&
port &&
privateKey &&
serverCaCert
) {
const tlsSocket = getSocket({
instanceInfo,
ephemeralCert,
host,
port,
privateKey,
serverCaCert,
instanceDnsName: dnsName,
serverName: instanceInfo.domainName || dnsName,
});
tlsSocket.once('error', () => {
cloudSqlInstance.forceRefresh();
});
tlsSocket.once('secureConnect', async () => {
cloudSqlInstance.setEstablishedConnection();
});

cloudSqlInstance.addSocket(tlsSocket);

return tlsSocket;
}

throw new CloudSQLConnectorError({
message: 'Invalid Cloud SQL Instance info',
code: 'EBADINSTANCEINFO',
});
}

const socket = new Socket();
socket.connect(tunnelPort, '127.0.0.1');

const cloudSqlInstance = instances.getInstance(opts);
socket.once('connect', () => {
cloudSqlInstance.setEstablishedConnection();
});
socket.once('error', () => {
cloudSqlInstance.forceRefresh();
});

cloudSqlInstance.addSocket(socket);

socket.connect = () => socket;

return socket;
},
};
}

return {
stream() {
const cloudSqlInstance = instances.getInstance(opts);
Expand Down
12 changes: 12 additions & 0 deletions src/ip-addresses.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ export enum IpAddressTypes {
PUBLIC = 'PUBLIC',
PRIVATE = 'PRIVATE',
PSC = 'PSC',
SQL_DATA = 'SQL_DATA',
}

export declare interface IpAddresses {
Expand Down Expand Up @@ -67,6 +68,17 @@ export function selectIpAddress(
return getPrivateIpAddress(ipAddresses);
case IpAddressTypes.PSC:
return getPSCIpAddress(ipAddresses);
case IpAddressTypes.SQL_DATA:
if (ipAddresses.public) {
return getPublicIpAddress(ipAddresses);
}
if (ipAddresses.private) {
return getPrivateIpAddress(ipAddresses);
}
if (ipAddresses.psc) {
return getPSCIpAddress(ipAddresses);
}
return '';
default:
throw new CloudSQLConnectorError({
message: 'Cannot connect to instance, it has no supported IP addresses',
Expand Down
Loading
Loading