1. If you haven't already, set up a Python Development Environment by following the [python setup guide](https://cloud.google.com/python/setup) and

[create a project](https://cloud.google.com/resource-manager/docs/creating-managing-projects#creating_a_project).

1. Create a 2nd Gen Cloud SQL Instance by following these

[instructions](https://cloud.google.com/sql/docs/mysql/create-instance). Note the connection string,

database user, and database password that you create.

1. Create a database for your application by following these

[instructions](https://cloud.google.com/sql/docs/mysql/create-manage-databases). Note the database

name.

1. Create a KMS key for your application by following these

[instructions](https://cloud.google.com/kms/docs/creating-keys). Copy the resource name of your

created key.

1. Create a service account with the 'Cloud SQL Client' permissions by following these

[instructions](https://cloud.google.com/sql/docs/mysql/connect-external-app#4_if_required_by_your_authentication_method_create_a_service_account).

Download a JSON key to use to authenticate your connection.

1. **macOS / Windows only**: Configure gRPC Root Certificates: On some platforms you may need to

accept the Google server certificates, see instructions for setting up

[root certs](https://github.com/googleapis/google-cloud-cpp/blob/master/google/cloud/bigtable/examples/README.md#configure-grpc-root-certificates).

To run this application locally, download and install the `cloud_sql_proxy` by

following the instructions

[here](https://cloud.google.com/sql/docs/mysql/sql-proxy#install).

Instructions are provided below for using the proxy with a TCP connection or a Unix Domain Socket.

On Linux or Mac OS you can use either option, but on Windows the proxy currently requires a TCP

connection.

To run the sample locally with a TCP connection, set environment variables and launch the proxy as

shown below.

Use these terminal commands to initialize environment variables:

export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service/account/key.json

export DB_HOST='127.0.0.1:3306'

export DB_USER='<DB_USER_NAME>'

export DB_PASS='<DB_PASSWORD>'

export DB_NAME='<DB_NAME>'

export GCP_KMS_URI='<GCP_KMS_URI>'

Note: Saving credentials in environment variables is convenient, but not secure - consider a more

secure solution such as [Secret Manager](https://cloud.google.com/secret-manager/docs/overview) to

help keep secrets safe.

Then use this command to launch the proxy in the background:

./cloud_sql_proxy -instances=<project-id>:<region>:<instance-name>=tcp:3306 -credential_file=$GOOGLE_APPLICATION_CREDENTIALS &

Use these PowerShell commands to initialize environment variables:

Note: Saving credentials in environment variables is convenient, but not secure - consider a more

secure solution such as [Secret Manager](https://cloud.google.com/secret-manager/docs/overview) to

help keep secrets safe.

Then use this command to launch the proxy in a separate PowerShell session:

NOTE: this option is currently only supported on Linux and Mac OS. Windows users should use the

[Launch proxy with TCP](#launch-proxy-with-tcp) option.

To use a Unix socket, you'll need to create a directory and give write access to the user running

the proxy. For example:

sudo mkdir /cloudsql

sudo chown -R $USER /cloudsql

You'll also need to initialize an environment variable containing the directory you just created:

export DB_SOCKET_DIR=/path/to/the/new/directory

Use these terminal commands to initialize other environment variables as well:

export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service/account/key.json

export INSTANCE_CONNECTION_NAME='<MY-PROJECT>:<INSTANCE-REGION>:<INSTANCE-NAME>'

export DB_USER='<DB_USER_NAME>'

export DB_PASS='<DB_PASSWORD>'

export DB_NAME='<DB_NAME>'

export GCP_KMS_URI='<GCP_KMS_URI>'

Note: Saving credentials in environment variables is convenient, but not secure - consider a more

secure solution such as [Secret Manager](https://cloud.google.com/secret-manager/docs/overview) to

help keep secrets safe.

Then use this command to launch the proxy in the background:

./cloud_sql_proxy -dir=$DB_SOCKET_DIR --instances=$INSTANCE_CONNECTION_NAME --credential_file=$GOOGLE_APPLICATION_CREDENTIALS &

Next, setup install the requirements into a virtual enviroment:

virtualenv --python python3 env

source env/bin/activate

pip install -r requirements.txt

Add new votes and the collected votes:

python snippets/query_and_decrypt_data.py

TEST_CONFIG_OVERRIDE = {

# You can opt out from the test for specific Python versions.

"ignored_versions": ["2.7", "3.6"],

# Old samples are opted out of enforcing Python type hints

# All new samples should feature them

"enforce_type_hints": True,

# An envvar key for determining the project id to use. Change it

# to 'BUILD_SPECIFIC_GCLOUD_PROJECT' if you want to opt in using a

# build specific Cloud project. You can also use your own string

# to use your own Cloud project.

"gcloud_project_env": "GOOGLE_CLOUD_PROJECT",

# 'gcloud_project_env': 'BUILD_SPECIFIC_GCLOUD_PROJECT',

# A dictionary you want to inject into your test. Don't put any

# secrets here. These values will override predefined values.

"envs": {},

}

pytest==6.2.2

import logging

import tink

from tink import aead

from tink.integration import gcpkms

logger = logging.getLogger(__name__)

def init_tink_env_aead(

key_uri: str,

credentials: str) -> tink.aead.KmsEnvelopeAead:

aead.register()

try:

gcp_client = gcpkms.GcpKmsClient(key_uri, credentials)

gcp_aead = gcp_client.get_aead(key_uri)

except tink.TinkError as e:

logger.error("Error initializing GCP client: %s", e)

raise e

# Create envelope AEAD primitive using AES256 GCM for encrypting the data

# This key should only be used for client-side encryption to ensure authenticity and integrity

# of data.

key_template = aead.aead_key_templates.AES256_GCM

env_aead = aead.KmsEnvelopeAead(key_template, gcp_aead)

print(f"Created envelope AEAD Primitive using KMS URI: {key_uri}")

return env_aead

import os

import pytest

from snippets.cloud_kms_env_aead import init_tink_env_aead

def setup() -> str:

kms_uri = "gcp-kms://" + os.environ["CLOUD_KMS_KEY"]

yield kms_uri

def test_cloud_kms_env_aead(

capsys: pytest.CaptureFixture, kms_uri: str) -> None:

credentials = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "")

# Create env_aead primitive

init_tink_env_aead(kms_uri, credentials)

captured = capsys.readouterr().out

assert f"Created envelope AEAD Primitive using KMS URI: {kms_uri}" in captured

import sqlalchemy

def init_tcp_connection_engine(

db_user: str, db_pass: str, db_name: str, db_host: str

) -> sqlalchemy.engine.base.Engine:

# Remember - storing secrets in plaintext is potentially unsafe. Consider using

# something like https://cloud.google.com/secret-manager/docs/overview to help keep

# secrets secret.

# Extract host and port from db_host

host_args = db_host.split(":")

db_hostname, db_port = host_args[0], int(host_args[1])

pool = sqlalchemy.create_engine(

# Equivalent URL:

# mysql+pymysql://<db_user>:<db_pass>@<db_host>:<db_port>/<db_name>

sqlalchemy.engine.url.URL(

drivername="mysql+pymysql",

username=db_user, # e.g. "my-database-user"

password=db_pass, # e.g. "my-database-password"

host=db_hostname, # e.g. "127.0.0.1"

port=db_port, # e.g. 3306

database=db_name, # e.g. "my-database-name"

),

)

print("Created TCP connection pool")

return pool

def init_unix_connection_engine(

db_user: str,

db_pass: str,

db_name: str,

cloud_sql_connection_name: str,

db_socket_dir: str,

) -> sqlalchemy.engine.base.Engine:

# Remember - storing secrets in plaintext is potentially unsafe. Consider using

# something like https://cloud.google.com/secret-manager/docs/overview to help keep

# secrets secret.

pool = sqlalchemy.create_engine(

# Equivalent URL:

# mysql+pymysql://<db_user>:<db_pass>@/<db_name>?unix_socket=<socket_path>/<cloud_sql_instance_name>

sqlalchemy.engine.url.URL(

drivername="mysql+pymysql",

username=db_user, # e.g. "my-database-user"

password=db_pass, # e.g. "my-database-password"

database=db_name, # e.g. "my-database-name"

query={"unix_socket": f"{db_socket_dir}/{cloud_sql_connection_name}"},

),

)

print("Created Unix socket connection pool")

return pool

def init_db(

db_user: str,

db_pass: str,

db_name: str,

table_name: str,

cloud_sql_connection_name: str = None,

db_socket_dir: str = None,

db_host: str = None,

) -> sqlalchemy.engine.base.Engine:

if db_host:

db = init_tcp_connection_engine(db_user, db_pass, db_name, db_host)

else:

db = init_unix_connection_engine(

db_user, db_pass, db_name, cloud_sql_connection_name, db_socket_dir

)

# Create tables (if they don't already exist)

with db.connect() as conn:

conn.execute(

f"CREATE TABLE IF NOT EXISTS {table_name} "

"( vote_id SERIAL NOT NULL, time_cast timestamp NOT NULL, "

"team CHAR(6) NOT NULL, voter_email VARBINARY(255), "

"PRIMARY KEY (vote_id) );"

)

print(f"Created table {table_name} in db {db_name}")

return db