使用 Security Command Center API 列出安全性发现结果

>

Security Command Center 发现结果模型可为组织资源的潜在安全风险建模。发现结果始终与 Security Command Center 中的特定资源相关。

本指南介绍了如何使用 Security Command Center 客户端库访问组织的发现结果。每个结果都属于一个来源。大多数检测器或发现结果提供程序在同一来源内生成发现结果。

准备工作

在设置来源之前，您需要完成以下操作：

页面大小

所有 Security Command Center list API 都是分页的。每个响应都会返回一个结果页面和一个返回下一页的令牌。页面大小可配置。默认的 pageSize 为 10，可设置为最小值 1 和最大值 1000。

发现结果保留

发现结果包含一系列 event_time 快照，用于捕获扫描过程中每次遇到关联漏洞或威胁时发现结果的状态和属性。

Security Command Center 从其 event_time 或事件发生的时间存储 13 个月的快照。13 个月后，发现结果快照及其数据会从 Security Command Center 数据库中删除，并且无法恢复。这会导致发现结果中的快照更少，导致无法查看发现结果及其随时间变化的方式。

只要发现结果至少有一个快照包含最近 13 个月的 event_time，就会持续保存在 Security Command Center 中。如需将发现结果及其所有数据保留更长时间，请将其导出到其他存储位置。如需相关说明，请参阅导出 Security Command Center 数据。

列出组织中的所有发现结果

gcloud

  # ORGANIZATION_ID=organization-id

  gcloud scc findings list $ORGANIZATION_ID

如需查看更多示例，请运行以下命令：

  gcloud scc findings list --help

Python

from google.cloud import securitycenter

# Create a client.
client = securitycenter.SecurityCenterClient()

# organization_id is the numeric ID of the organization. e.g.:
# organization_id = "111122222444"
org_name = "organizations/{org_id}".format(org_id=organization_id)
# The "sources/-" suffix lists findings across all sources.  You
# also use a specific source_name instead.
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):
    print(
        "{}: name: {} resource: {}".format(
            i, finding_result.finding.name, finding_result.finding.resource_name
        )
    )

Java

static ImmutableList<ListFindingsResult> listAllFindings(OrganizationName organizationName) {
  try (SecurityCenterClient client = SecurityCenterClient.create()) {
    // OrganizationName organizationName = OrganizationName.of(/*organizationId=*/"123234324");
    // "-" Indicates listing across all sources.
    SourceName sourceName = SourceName.of(organizationName.getOrganization(), "-");

    ListFindingsRequest.Builder request =
        ListFindingsRequest.newBuilder().setParent(sourceName.toString());

    // Call the API.
    ListFindingsPagedResponse response = client.listFindings(request.build());

    // This creates one list for all findings.  If your organization has a large number of
    // findings this can cause out of memory issues.  You can process them in incrementally
    // by returning the Iterable returned response.iterateAll() directly.
    ImmutableList<ListFindingsResult> results = ImmutableList.copyOf(response.iterateAll());
    System.out.println("Findings:");
    System.out.println(results);
    return results;
  } catch (IOException e) {
    throw new RuntimeException("Couldn't create client.", e);
  }
}

Go

import (
	"context"
	"fmt"
	"io"

	securitycenter "cloud.google.com/go/securitycenter/apiv1"
	"google.golang.org/api/iterator"
	securitycenterpb "google.golang.org/genproto/googleapis/cloud/securitycenter/v1"
)

// listFindings prints all findings in orgID to w.  orgID is the numeric
// identifier of the organization.
func listFindings(w io.Writer, orgID string) error {
	// orgID := "12321311"
	// Instantiate a context and a security service client to make API calls.
	ctx := context.Background()
	client, err := securitycenter.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("securitycenter.NewClient: %v", err)
	}
	defer client.Close() // Closing the client safely cleans up background resources.

	req := &securitycenterpb.ListFindingsRequest{
		// List findings across all sources.
		Parent: fmt.Sprintf("organizations/%s/sources/-", orgID),
	}
	it := client.ListFindings(ctx, req)
	for {
		result, err := it.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("it.Next: %v", err)
		}
		finding := result.Finding
		fmt.Fprintf(w, "Finding Name: %s, ", finding.Name)
		fmt.Fprintf(w, "Resource Name %s, ", finding.ResourceName)
		fmt.Fprintf(w, "Category: %s\n", finding.Category)
	}
	return nil
}

Node.js

// Imports the Google Cloud client library.
const {SecurityCenterClient} = require('@google-cloud/security-center');

// Creates a new client.
const client = new SecurityCenterClient();
//  organizationId is the numeric ID of the organization.
/*
 * TODO(developer): Uncomment the following lines
 */
// const organizationId = "1234567777";

async function listAllFindings() {
  const [response] = await client.listFindings({
    // List findings across all sources.
    parent: `organizations/${organizationId}/sources/-`,
  });
  let count = 0;
  Array.from(response).forEach(result =>
    console.log(
      `${++count} ${result.finding.name} ${result.finding.resourceName}`
    )
  );
}
listAllFindings();

每个发现结果的输出如下所示：

finding:
  category: 'Persistence: IAM Anomalous Grant'
  createTime: '2020-09-13T12:30:10.248Z'
  eventTime: '2020-09-13T12:30:09.528Z'
  name: organizations/organization-id/sources/source-id/findings/finding-id
  parent: organizations/organization-id/sources/source-id
  resourceName: //cloudresourcemanager.googleapis.com/projects/project-number
  securityMarks:
    name: organizations/organization-id/sources/source-id/findings/finding-id/securityMarks
  sourceProperties:
    affectedResources:
    - gcpResourceName: //cloudresourcemanager.googleapis.com/projects/project-number
    detectionCategory:
      indicator: audit_log
      ruleName: iam_anomalous_grant
      subRuleName: service_account_granted_sensitive_role_to_member
      technique: persistence
    detectionPriority: HIGH
    evidence:
    - sourceLogId:
        insertId: insert-id
        projectId: project-id
        timestamp:
          nanos: 183000000
          seconds: '1600000206'
    findingId: finding-id
    properties:
      sensitiveRoleGrant:
        bindingDeltas:
        - action: ADD
          member: user:user-email@gmail.com
          role: roles/editor
        members:
        - user:user-email@gmail.com
        principalEmail: principal-email.iam.gserviceaccount.com
    sourceId:
      customerOrganizationNumber: 'organization-id'
      projectNumber: 'project-number'
  state: ACTIVE
resource:
  name: //cloudresourcemanager.googleapis.com/projects/project-number
  parentDisplayName: organization-name
  parentName: //cloudresourcemanager.googleapis.com/folders/folder-id
  projectDisplayName: project-id
  projectName: //cloudresourcemanager.googleapis.com/projects/project-number

过滤发现结果

一个组织可能会有许多发现结果。上面的示例未使用过滤条件，因此会返回所有发现结果记录。借助 Security Command Center，您可以使用发现结果过滤条件来获取所需发现结果的相关信息，并将父级限制为特定来源。

发现结果的过滤条件与 SQL 语句中的“where”子句类似，但列除外，它们会应用于 API 返回的对象。

以下示例仅列出类别为“MEDIUM_RISK_ONE”的发现结果。具体类别可能发生变化，您应查阅发现结果服务提供商的文档以确定其使用的类别。

gcloud

  # ORGANIZATION_ID=organization-id
  # SOURCE_ID="source-id"
  FILTER="category=\"MEDIUM_RISK_ONE\""

  gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID --filter="$FILTER"

如需查看更多示例，请运行以下命令：

  gcloud scc findings list --help

Python

from google.cloud import securitycenter

# Create a new client.
client = securitycenter.SecurityCenterClient()

# source_name is the resource path for a source that has been
# created previously (you can use list_sources to find a specific one).
# Its format is:
# source_name = "organizations/{organization_id}/sources/{source_id}"
# e.g.:
# source_name = "organizations/111122222444/sources/1234"
# You an also use a wild-card "-" for all sources:
#   source_name = "organizations/111122222444/sources/-"
finding_result_iterator = client.list_findings(
    request={"parent": source_name, "filter": 'category="MEDIUM_RISK_ONE"'}
)
# Iterate an print all finding names and the resource they are
# in reference to.
for i, finding_result in enumerate(finding_result_iterator):
    print(
        "{}: name: {} resource: {}".format(
            i, finding_result.finding.name, finding_result.finding.resource_name
        )
    )

Java

static ImmutableList<ListFindingsResult> listFilteredFindings(SourceName sourceName) {
  try (SecurityCenterClient client = SecurityCenterClient.create()) {
    // SourceName sourceName = SourceName.of(/*organizationId=*/"123234324",
    // /*sourceId=*/"423432321");

    // Create filter to category of MEDIUM_RISK_ONE
    String filter = "category=\"MEDIUM_RISK_ONE\"";

    ListFindingsRequest.Builder request =
        ListFindingsRequest.newBuilder().setParent(sourceName.toString()).setFilter(filter);

    // Call the API.
    ListFindingsPagedResponse response = client.listFindings(request.build());

    // This creates one list for all findings.  If your organization has a large number of
    // findings this can cause out of memory issues.  You can process them in incrementally
    // by returning the Iterable returned response.iterateAll() directly.
    ImmutableList<ListFindingsResult> results = ImmutableList.copyOf(response.iterateAll());
    System.out.println("Findings:");
    System.out.println(results);
    return results;
  } catch (IOException e) {
    throw new RuntimeException("Couldn't create client.", e);
  }
}

Go

import (
	"context"
	"fmt"
	"io"

	securitycenter "cloud.google.com/go/securitycenter/apiv1"
	"google.golang.org/api/iterator"
	securitycenterpb "google.golang.org/genproto/googleapis/cloud/securitycenter/v1"
)

// listFilteredFindings prints findings with category 'MEDIUM_RISK_ONE' for a
// specific source to w. sourceName is the full resource name of the source
// to search for findings under.
func listFilteredFindings(w io.Writer, sourceName string) error {
	// Specific source.
	// sourceName := "organizations/111122222444/sources/1234"
	// All sources.
	// sourceName := "organizations/111122222444/sources/-"
	// Instantiate a context and a security service client to make API calls.
	ctx := context.Background()
	client, err := securitycenter.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("securitycenter.NewClient: %v", err)
	}
	defer client.Close() // Closing the client safely cleans up background resources.

	req := &securitycenterpb.ListFindingsRequest{
		Parent: sourceName,
		Filter: `category="MEDIUM_RISK_ONE"`,
	}
	it := client.ListFindings(ctx, req)
	for {
		result, err := it.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("it.Next: %v", err)
		}
		finding := result.Finding
		fmt.Fprintf(w, "Finding Name: %s, ", finding.Name)
		fmt.Fprintf(w, "Resource Name %s, ", finding.ResourceName)
		fmt.Fprintf(w, "Category: %s\n", finding.Category)
	}
	return nil
}

Node.js

// Imports the Google Cloud client library.
const {SecurityCenterClient} = require('@google-cloud/security-center');

// Creates a new client.
const client = new SecurityCenterClient();
//  sourceName is the full resource path of the source to search for
//  findings.
/*
 * TODO(developer): Uncomment the following lines
 */
// const sourceName = "organizations/111122222444/sources/1234";

async function listFilteredFindings() {
  const [response] = await client.listFindings({
    // List findings across all sources.
    parent: sourceName,
    filter: 'category="MEDIUM_RISK_ONE"',
  });
  let count = 0;
  Array.from(response).forEach(result =>
    console.log(
      `${++count} ${result.finding.name} ${result.finding.resourceName}`
    )
  );
}
listFilteredFindings();

Security Command Center 还支持完整的 JSON 数组和对象作为潜在属性类型。您可以按以下条件进行过滤：

  • 数组元素
  • 对象中部分字符串匹配的完整 JSON 对象
  • JSON 对象子字段

子字段必须是数字、字符串或布尔值，并且必须使用以下运算符：

  • 字符串
    • 完全相等运算符 =
    • 部分字符串与 : 匹配
  • 数字
    • 不等式 <、>、<=、>=
    • 相等 =
  • 布尔值
    • 相等 =

本页面后面的示例假定以下 JSON 对象是发现结果的源属性：

{
    "outer_object": {
        "middle_object": {
            "deeply_nested_object": {
                "x": 123,
            },
            "y": "some-string-value",
        },
        "z": "some-other-string-value",
        "u": ["list-element-1", "list-element-2", "list-element-3", ],
    },
}

过滤发现结果示例

在此示例中，上一个 JSON 对象是名为 my_property 的发现结果的来源属性。以下示例展示了如何将对象作为属性的发现结果进行查询。您还可以在查询中使用 AND 和 OR 将这些过滤条件与其他过滤条件一起使用。

# ORGANIZATION_ID=organization-id
# SOURCE_ID="source-id"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.middle_object.deeply_nested_object.x = 123"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.middle_object.y = \"some-string-value\""

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.middle_object.y : \"string-value\""

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.z = \"some-other-string-value\""

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.z : \"other-string-value\""

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.u : \"list-element-1\""

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.u : \"list-element-2\""

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --filter="source_properties.my_property.outer_object.u : \"list-element-3\""

对发现结果进行排序的示例

您可以按原始类型（字符串，数字和布尔值）的严格子字段对结果进行排序。在此示例中，上一个 JSON 对象是名为 my_property 的发现结果的来源属性。以下示例包含用于对发现结果字段进行排序的查询：

# ORGANIZATION_ID=organization-id
# SOURCE_ID="source-id"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --order-by="source_properties.my_property.outer_object.middle_object.deeply_nested_object.x DESC"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --order-by="source_properties.my_property.outer_object.middle_object.deeply_nested_object.x"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --order-by="source_properties.my_property.outer_object.middle_object.y DESC"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --order-by="source_properties.my_property.outer_object.middle_object.y"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --order-by="source_properties.my_property.outer_object.z DESC"

gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID \
  --order-by="source_properties.my_property.outer_object.z"

时间点查询

Security Command Center 可让您列出 13 个月内特定快照时间的发现结果。如需了解详情，请参阅发现结果保留。

gcloud

  # ORGANIZATION_ID=organization-id
  # SOURCE_ID="source-id"
  # READ_TIME follows the format YYYY-MM-DDThh:mm:ss.ffffffZ
  READ_TIME=2019-02-28T07:00:06.861Z

  gcloud scc findings list $ORGANIZATION_ID --source=$SOURCE_ID --read-time=$READ_TIME

如需查看更多示例，请运行以下命令：

  gcloud scc findings list --help

Python

from google.cloud import securitycenter
from datetime import timedelta, datetime

# Create a new client.
client = securitycenter.SecurityCenterClient()

# source_name is the resource path for a source that has been
# created previously (you can use list_sources to find a specific one).
# Its format is:
# source_name = "organizations/{organization_id}/sources/{source_id}"
# e.g.:
# source_name = "organizations/111122222444/sources/1234"
# You an also use a wild-card "-" for all sources:
#   source_name = "organizations/111122222444/sources/-"
five_days_ago = str(datetime.now() - timedelta(days=5))

finding_result_iterator = client.list_findings(
    request={"parent": source_name, "filter": five_days_ago}
)
for i, finding_result in enumerate(finding_result_iterator):
    print(
        "{}: name: {} resource: {}".format(
            i, finding_result.finding.name, finding_result.finding.resource_name
        )
    )

Java

static ImmutableList<ListFindingsResult> listFindingsAtTime(SourceName sourceName) {
  try (SecurityCenterClient client = SecurityCenterClient.create()) {
    // SourceName sourceName = SourceName.of(/*organizationId=*/"123234324",
    // /*sourceId=*/"423432321");

    // 5 days ago
    Instant fiveDaysAgo = Instant.now().minus(Duration.ofDays(5));

    ListFindingsRequest.Builder request =
        ListFindingsRequest.newBuilder()
            .setParent(sourceName.toString())
            .setReadTime(
                Timestamp.newBuilder()
                    .setSeconds(fiveDaysAgo.getEpochSecond())
                    .setNanos(fiveDaysAgo.getNano()));

    // Call the API.
    ListFindingsPagedResponse response = client.listFindings(request.build());

    // This creates one list for all findings.  If your organization has a large number of
    // findings this can cause out of memory issues.  You can process them in incrementally
    // by returning the Iterable returned response.iterateAll() directly.
    ImmutableList<ListFindingsResult> results = ImmutableList.copyOf(response.iterateAll());
    System.out.println("Findings:");
    System.out.println(results);
    return results;
  } catch (IOException e) {
    throw new RuntimeException("Couldn't create client.", e);
  }
}

Go

import (
	"context"
	"fmt"
	"io"
	"time"

	securitycenter "cloud.google.com/go/securitycenter/apiv1"
	"github.com/golang/protobuf/ptypes"
	"google.golang.org/api/iterator"
	securitycenterpb "google.golang.org/genproto/googleapis/cloud/securitycenter/v1"
)

// listFindingsAtTime prints findings that where present for a specific source
// as of five days ago to w. sourceName is the full resource name of the
// source to search for findings under.
func listFindingsAtTime(w io.Writer, sourceName string) error {
	// Specific source.
	// sourceName := "organizations/111122222444/sources/1234"
	// All sources.
	// sourceName := "organizations/111122222444/sources/-"
	ctx := context.Background()
	client, err := securitycenter.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("securitycenter.NewClient: %v", err)
	}
	defer client.Close() // Closing the client safely cleans up background resources.
	fiveDaysAgo, err := ptypes.TimestampProto(time.Now().AddDate(0, 0, -5))
	if err != nil {
		return fmt.Errorf("Error converting five days ago: %v", err)
	}

	req := &securitycenterpb.ListFindingsRequest{
		Parent:   sourceName,
		ReadTime: fiveDaysAgo,
	}
	it := client.ListFindings(ctx, req)
	for {
		result, err := it.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return fmt.Errorf("it.Next: %v", err)
		}
		finding := result.Finding
		fmt.Fprintf(w, "Finding Name: %s, ", finding.Name)
		fmt.Fprintf(w, "Resource Name %s, ", finding.ResourceName)
		fmt.Fprintf(w, "Category: %s\n", finding.Category)
	}
	return nil
}

Node.js

// Imports the Google Cloud client library.
const {SecurityCenterClient} = require('@google-cloud/security-center');

// Creates a new client.
const client = new SecurityCenterClient();
// sourceName is the fully qualified source name to search for findings
// under.
/*
 * TODO(developer): Uncomment the following lines
 */
// const sourceName = "organizations/111122222444/sources/1234";

const fiveDaysAgo = new Date();
fiveDaysAgo.setDate(fiveDaysAgo.getDate() - 5);

async function listFindingsAtTime() {
  const [response] = await client.listFindings({
    // List findings across all sources.
    parent: sourceName,
    readTime: {
      seconds: Math.floor(fiveDaysAgo.getTime() / 1000),
      nanos: (fiveDaysAgo.getTime() % 1000) * 1e6,
    },
  });
  let count = 0;
  Array.from(response).forEach(result =>
    console.log(
      `${++count} ${result.finding.name} ${result.finding.resourceName}`
    )
  );
}
listFindingsAtTime();

FilterExamples

以下是一些其他有用的发现结果过滤器。

在某个时间点之后出现的发现结果

以下示例过滤条件将匹配最近 2019 年 6 月 5 日星期三下午 10:12:05 （格林尼治标准时间）之后的最新发现结果。借助 event_time 过滤器，您可以使用以下格式和类型表示时间：

  • 以整数字面量表示的 Unix 时间（以毫秒为单位）

    "event_time > 1559772725000"

  • 作为字符串字面量的 RFC 3339

    "event_time > \"2019-06-05T22:34:40+00:00\""