Secure access to Google Cloud Resources
When your organization is growing quickly their use of public cloud, security should not be left as an afterthought.
One of the most important aspect is the ServiceAccount (we call it SA) and the keys that are attached to it.
The SA is not the problem here, as Google recommends replacing default SA’s with specific ones and separate them for each services used in GCP.
The biggest problem is the external KEY, we can call it exfiltration of data, we can have multiple people sharing the same key, let’s face it : it’s out of control !!!
Most of the time people just create the JSON Key because it’s suggested by default, and because in the past it was mandatory on many many API’s.
Now when you have 2 differents services that need to talk to one another on GCP you just have to use the get_application_default() on your code with the GCP service called (Just don’t forget to use a specific service account different for each component, for example for a specific Cloud Function with the least privileges principle)
from oauth2client.client import GoogleCredentials
CREDENTIALS = GoogleCredentials.get_application_default()On top of that there is a feature called Organizational Policy, where you have many differents possibilities
Today we chose not to deploy this policy :
Disable service account creation
This Boolean constraint disables the creation of service accounts where this constraint is set to ‘True’. By default, service accounts can be created by users based on their Cloud IAM roles and permissions.
BUT to we chose deploy this policy :
Disable service account key creation
This Boolean constraint disables the creation of service account external keys where this constraint is set to ‘True’. By default, service account external keys can be created by users based on their Cloud IAM roles and permissions.
Because for us as i’ve said before, SA is not a problem, but having to many external Keys is a real problem !
So now that it’s enabled when someone tries to generate a KEY they got this message :
Create failed. Error 400: Key creation is not allowed on this service account.That’s awesome !
But…sometime it’s impossible to not accept a key, because some external tools require that as an authentication method (we don’t mention them but imagine another Cloud Provider trying to communicate with GCP for example)
So we decided to provide a solution for people who really need an external Key.
A form to request a key, and after approvalwe want everything to be done automatically because we don’t really like to spend all day creating external key manually….
Here is what we will do :
Now let me give you the sample code of each step (currently i’m doing every step in just one main function)
I’m mostly using the Discovery API because the libraries are not equivalent, some times the features that we need are not yet available.
Step 1
A message sent to Cloud Pub/Sub triggers a Cloud Function
No code needed here because because it’s automatically done by Google
Step 2
The org policy is temporary disabled on the specified project
from googleapiclient import discoverySERVICE_RESOURCE_MANAGER = discovery.build(‘cloudresourcemanager’, ‘v1’,
credentials=CREDENTIALS)
resource = ‘projects/’+project_id
disable_org_policy_request_body = {
“policy”: {
“constraint”: "constraints/iam.disableServiceAccountKeyCreation",
“booleanPolicy”:
{
“enforced”: False}}}def disable_policy_operator(resource, body):request = SERVICE_RESOURCE_MANAGER.projects().setOrgPolicy(resource=resource, body=disable_org_policy_request_body)
response = request.execute()
return response
Step 3
The different required API’s are enabled :
- cloudapis.googleapis.com to intialize default APIs for a new project
- cloudbilling.googleapis.com for the next step
- billingbudgets.googleapis.com for the next step
- serviceusage.googleapis.com for enabling APIs
- cloudresourcemanager.googleapis.com
- iam.googleapis.com for the IAM Service Account KEY generation
from googleapiclient import discoverySERVICE_USAGE_MANAGER = discovery.build('serviceusage', 'v1',
credentials=CREDENTIALS)APIS = [
‘cloudapis.googleapis.com’,
‘cloudbilling.googleapis.com’,
‘serviceusage.googleapis.com’,
‘cloudresourcemanager.googleapis.com’,
‘iam.googleapis.com’,
‘billingbudgets.googleapis.com’]project_id = “projects/” + project_name
body = {“serviceIds”: batch_apis}response = enable_batch_apis_operator(project_id, body)def enable_batch_apis_operator(project_id, body):request = SERVICE_USAGE_MANAGER.services().batchEnable(parent=project_id, body=body)
response = request.execute()
return response
Step 4
The billing API is called to know if the project is linked to a billing or not (otherwise we add it and create a small budget alert using an API too)
from googleapiclient import discoverySERVICE_CLOUD_BILLING = discovery.build('cloudbilling', 'v1', credentials=CREDENTIALS)request = SERVICE_CLOUD_BILLING.projects().getBillingInfo(name="projects/"{project_id})
response = request.execute()
return response
Step 5
We enable the SecretManager API (it’s mandatory to have a billing account before enabling it) secretmanager.googleapis.com
from googleapiclient import discoverySERVICE_USAGE_MANAGER = discovery.build('serviceusage', 'v1',credentials=CREDENTIALS)body = “projects/” + project_name + “/services/secretmanager.googleapis.com”request = SERVICE_USAGE_MANAGER.services().enable(name=body)
response = request.execute()
Step 6
We generate the JSON KEY without storing it anywhere
from googleapiclient import discoverySERVICE_ACCOUNT_MANAGER = discovery.build('iam', 'v1', credentials=CREDENTIALS)request = SERVICE_ACCOUNT_MANAGER.projects().serviceAccounts().keys().create(name=’projects/’+project_id+’/serviceAccounts/’ + service_account_email, body={})response = request.execute()
return response[‘privateKeyData’]
Step 7
We send it to SecretManager and :
- First, initate the connection to the SecretManager
from google.cloud import secretmanager
SECRET_MANAGER = secretmanager.SecretManagerServiceClient()- Check if a Secret for this SA already exists
def check_secret_version_operator(secret_id, project):name = SECRET_MANAGER.secret_version_path(project, secret_id, 'latest')
response = SECRET_MANAGER.access_secret_version(name)
return response
- if it already exists we simply just generate a new version
def create_secret_version_operator(secret_id, key, project):key = base64.b64decode(key)
name = SECRET_MANAGER.secret_version_path(project, secret_id, 'latest')
response = SECRET_MANAGER.access_secret_version(name)
logging.info(response)
try:
name = SECRET_MANAGER.secret_version_path(project, secret_id, 'v1')
response = SECRET_MANAGER.access_secret_version(name)
logging.info(response)except:
parent = SECRET_MANAGER.secret_path(project, secret_id)
request = SECRET_MANAGER.add_secret_version(parent, {'data': key})
response = SECRET_MANAGER.access_secret_version(request.name)
payload = response.payload.data.decode('UTF-8')
logging.info('Plaintext: {}'.format(payload))
logging.info('Created secret: {}'.format(response.name))
return response
- If not we create a Secret with the first version :
def create_secret_operator(project_id, secret_id):parent = SECRET_MANAGER.project_path(project_id)
secret = SECRET_MANAGER.create_secret(parent, secret_id, {'replication': {'automatic': {},},})
return secret_id
def create_secret_version_operator(secret_id, key, project):key = base64.b64decode(key)
name = SECRET_MANAGER.secret_version_path(project, secret_id, 'latest')
response = SECRET_MANAGER.access_secret_version(name)
logging.info(response)try:
name = SECRET_MANAGER.secret_version_path(project, secret_id, 'v1')
response = SECRET_MANAGER.access_secret_version(name)
logging.info(response)except:
parent = SECRET_MANAGER.secret_path(project, secret_id)
request = SECRET_MANAGER.add_secret_version(parent, {'data': key})
response = SECRET_MANAGER.access_secret_version(request.name)
payload = response.payload.data.decode('UTF-8')
logging.info('Plaintext: {}'.format(payload))
logging.info('Created secret: {}'.format(response.name))
return response
Step 8
We re-enable the policy on the project
from googleapiclient import discoverySERVICE_RESOURCE_MANAGER = discovery.build('cloudresourcemanager', 'v1',credentials=CREDENTIALS)resource = ‘projects/’+project_name
set_org_policy_request_body = {“constraint”: "constraints/iam.disableServiceAccountKeyCreation"}request = SERVICE_RESOURCE_MANAGER.projects().clearOrgPolicy(resource=resource, body=set_org_policy_request_body)response = request.execute()
We return the gcloud command to the user to get his key from Secret Manager
Every step is monitored, with Error Reporting, with just this sample of code :
from google.cloud import error_reporting
client = error_reporting.Client()
try:
raise RuntimeError('I failed you')
except RuntimeError:
client.report_exception()And we send an email (with SendGRID) to the team administrator if a step failed to be instantly informed :
def send_email(project_id_target, owner, status, step):sg = SendGridAPIClient("your_sendgrid_api_key")subject=f”Google Cloud Service Account JSON Key Creation”if status == ‘ERROR’:subject=f”Google Cloud Service Account JSON Key Creation Error, {step}”html_content = f”””<p>Hello, the project named {project_id_target},</p><p>with {owner} as owner</p>“””message = Mail(to_emails= email_var(),from_email=Email(email_var(), “Google Cloud Team”),subject=subject,html_content=html_content)try:response = sg.send(message)return f”email.status_code={response.status_code}”except HTTPError as e:return e.message
The next steps will be to offer a rotation of every key that we give and clean up the old ones… See you on the next episode !