Search code examples
pythongoogle-cloud-platformgoogle-bigquerysecuritycentercloud-security

How To Export GCP Security Command Center Findings To BigQuery?


Similar to this: How to export GCP's Security Center Assets to a Cloud Storage via cloud Function?

I need to export the Findings as seen in the Security Command Center to BigQuery so we can easily filter the data we need and generate custom reports.

Using this documentation as an example (https://cloud.google.com/security-command-center/docs/how-to-api-list-findings#python), I wrote the following:

from google.cloud import securitycenter
from google.cloud import bigquery

JSONPath = "Path to JSON File For Service Account"
client = securitycenter.SecurityCenterClient().from_service_account_json(JSONPath)
BQclient = bigquery.Client().from_service_account_json(JSONPath)
table_id = "project.security_center.assets"
org_name = "organizations/1234567891011"
all_sources = "{org_name}/sources/-".format(org_name=org_name)
finding_result_iterator = client.list_findings(request={"parent": all_sources})
for i, finding_result in enumerate(finding_result_iterator):
    errors = BQclient.insert_rows_json(table_id, finding_result)
    if errors == []:
        print("New rows have been added.")
    else:
        print("Encountered errors while inserting rows: {}".format(errors))

However, that then gave me the error:

"json_rows argument should be a sequence of dicts".

Any help with this would be greatly appreciated :)


Solution

  • I managed to sort this by writing:

    for i, finding_result in enumerate(finding_result_iterator):
    rows_to_insert = [
        {u"category": finding_result.finding.category, u"name": finding_result.finding.name, u"project": finding_result.resource.project_display_name, u"external_uri": finding_result.finding.external_uri},
    ]