Cloud Asset Inventory: Truy vấn trạng thái tài nguyên & tuân thủ

Vì sao cần Cloud Asset Inventory

Khi phân cấp mở rộng, rất khó để duy trì cái nhìn chính xác về toàn bộ resources:

Organization với 1000 projects, 100,000 resources:
- VM nào có external IP? (vi phạm tuân thủ nếu không mong đợi)
- Ai đang có role Owner? (kiểm toán bảo mật)
- Resource nào đang ở trạng thái soft-delete? (theo dõi quota)
- Resource Compute Engine nào không tuân thủ policy bảo mật?
- Trạng thái hiện tại của resource X là gì? (khắc phục sự cố)

Cách làm thủ công không hiệu quả:

• Truy vấn từng service riêng lẻ (chậm, thiếu sót)
• Chạy lệnh gcloud (không mở rộng tốt)
• Phân tích audit logs (phản ứng sau sự việc, không chủ động)

Cloud Asset Inventory cung cấp:

• Một view thống nhất của toàn bộ resources (200+ loại resource)
• API có thể truy vấn với hỗ trợ filtering/biểu thức
• Trạng thái resource thời gian thực (live query) + lịch sử (qua export)
• Công cụ tuân thủ (Policy Analyzer, tích hợp Security Command Center)

Kiến trúc

Cloud Asset Inventory duy trì hai kho dữ liệu:

1. Metadata tài sản thời gian thực

Organization → Folder → Project → Resources
     ↓
Cloud Asset Inventory (cơ sở dữ liệu đã index)
     ↓
Query API

Đặc điểm:

• Trạng thái resource trực tiếp
• Có thể truy vấn trong vài giây
• Lưu trữ theo region (phụ thuộc vào vị trí asset)
• Có đầy đủ ngữ cảnh phân cấp

2. Export lịch sử

Dữ liệu thời gian thực → Export vào BigQuery / Cloud Storage
     ↓
Phân tích lịch sử
Phát hiện xu hướng
Kiểm toán tuân thủ

Đặc điểm:

• Lưu đầy đủ lịch sử
• Truy vấn được bằng BigQuery SQL
• Có chính sách retention tùy chỉnh
• Chi phí: Storage + BigQuery query

Mẫu truy vấn

Mẫu 1: Tìm tất cả resource theo loại

# Tìm tất cả Compute Engine instances
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID \
  --asset-types=compute.googleapis.com/Instance

# Tìm tất cả Cloud Storage buckets
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID \
  --asset-types=storage.googleapis.com/Bucket

# Tìm nhiều loại cùng lúc
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID \
  --asset-types=compute.googleapis.com/Instance,storage.googleapis.com/Bucket

Mẫu 2: Lọc theo thuộc tính resource

# Tìm VM có external IP (rủi ro bảo mật tiềm năng)
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID \
  --asset-types=compute.googleapis.com/Instance \
  --query="networkInterfaces.accessConfigs.natIP:*"

# Tìm resource trong project cụ thể
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID \
  --query="project:my-project-prod"

# Tìm resource theo location
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID \
  --query="location:us-central1"

Mẫu 3: Kiểm toán tuân thủ

from google.cloud import asset_v1

def audit_external_ips(org_id):
    """Tìm tất cả VM có external IP không mong đợi"""
    
    client = asset_v1.AssetServiceClient()
    
    # Truy vấn VM có external IP
    query = "networkInterfaces.accessConfigs.natIP:*"
    
    request = asset_v1.SearchAllResourcesRequest(
        scope=f"organizations/{org_id}",
        asset_types=["compute.googleapis.com/Instance"],
        query=query
    )
    
    findings = []
    for result in client.search_all_resources(request=request):
        # Kiểm tra external IP có được phép hay không
        if not is_external_ip_allowed(result):
            findings.append({
                "resource": result.name,
                "project": result.project,
                "issue": "Unexpected external IP"
            })
    
    return findings

def audit_owner_role(org_id):
    """Tìm tất cả user/service account có role Owner"""
    
    client = asset_v1.AssetServiceClient()
    
    # Truy vấn IAM bindings
    request = asset_v1.SearchAllIamPoliciesRequest(
        scope=f"organizations/{org_id}",
        query='roles/owner'
    )
    
    findings = []
    for result in client.search_all_iam_policies(request=request):
        for binding in result.bindings:
            if binding.role == "roles/owner":
                for member in binding.members:
                    findings.append({
                        "resource": result.resource,
                        "member": member,
                        "role": binding.role
                    })
    
    return findings

Phát hiện drift

Cloud Asset Inventory cho phép kiểm tra tuân thủ tự động:

def detect_resource_drift(resource_id, expected_config):
    """Phát hiện resource có lệch khỏi trạng thái kỳ vọng hay không"""
    
    from google.cloud import asset_v1
    
    client = asset_v1.AssetServiceClient()
    
    # Lấy trạng thái hiện tại của asset
    response = client.list_assets(
        parent=f"projects/PROJECT_ID",
        asset_types=[resource_type]
    )
    
    actual_config = None
    for asset in response:
        if asset.name == resource_id:
            actual_config = asset.resource
            break
    
    if not actual_config:
        return {"status": "MISSING", "drift": expected_config}
    
    # So sánh các field
    drifts = {}
    for key, expected_value in expected_config.items():
        actual_value = actual_config.get(key)
        if actual_value != expected_value:
            drifts[key] = {
                "expected": expected_value,
                "actual": actual_value
            }
    
    if drifts:
        return {"status": "DRIFTED", "drifts": drifts}
    else:
        return {"status": "COMPLIANT"}

Kiểm toán IAM Policy

# Tìm tất cả service account có role cụ thể
gcloud asset search-all-iam-policies \
  --scope=organizations/ORG_ID \
  --query='roles/serviceaccountUser'

# Tìm quyền của một member cụ thể
gcloud asset search-all-iam-policies \
  --scope=organizations/ORG_ID \
  --query='members:user@example.com'

# Export tất cả IAM policies sang BigQuery
gcloud asset export --project=EXPORT_PROJECT \
  --bucket=gs://asset-export-bucket \
  --asset-types=cloudresourcemanager.googleapis.com/Project

Phân tích lịch sử qua BigQuery

-- Truy vấn dữ liệu asset lịch sử
-- (sau khi export Cloud Asset Inventory sang BigQuery)

SELECT
  TIMESTAMP(DATE(timestamp)) as date,
  asset_type,
  COUNT(*) as resource_count
FROM `project.dataset.cloudresourcemanager_assets_YYYYMM`
WHERE asset_type = 'compute.googleapis.com/Instance'
GROUP BY date, asset_type
ORDER BY date DESC;

-- Phát hiện resource mới tạo
SELECT
  name,
  display_name,
  create_time
FROM `project.dataset.compute_instances_YYYYMM`
WHERE DATE(create_time) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
ORDER BY create_time DESC;

-- Tìm resource thiếu label bắt buộc
SELECT
  name,
  display_name,
  STRUCT(
    JSON_EXTRACT_SCALAR(resource.data, '$.labels.environment') as environment,
    JSON_EXTRACT_SCALAR(resource.data, '$.labels.team') as team
  ) as labels
FROM `project.dataset.compute_instances_YYYYMM`
WHERE DATE(timestamp) = CURRENT_DATE()
  AND (
    JSON_EXTRACT_SCALAR(resource.data, '$.labels.environment') IS NULL
    OR JSON_EXTRACT_SCALAR(resource.data, '$.labels.team') IS NULL
  );

Tích hợp Security Command Center

from google.cloud import securitycenter_v1

def create_compliance_finding(org_id, resource_name, issue):
    """Tạo finding Security Command Center từ kiểm tra tuân thủ"""
    
    client = securitycenter_v1.SecurityCenterClient()
    
    parent = f"organizations/{org_id}"
    source_name = f"{parent}/sources/FINDING_SOURCE_ID"
    
    finding = securitycenter_v1.Finding(
        state=securitycenter_v1.Finding.State.ACTIVE,
        resource_name=resource_name,
        finding_class=securitycenter_v1.Finding.FindingClass.MISCONFIGURATION,
        severity=securitycenter_v1.Finding.Severity.HIGH,
        source_properties={
            "issue": issue,
            "recommendation": "Fix configuration"
        }
    )
    
    client.create_finding(
        parent=source_name,
        finding=finding
    )

Cân nhắc về khả năng mở rộng

Truy vấn quy mô lớn

# Với org lớn, truy vấn có thể mất thời gian
# Hãy dùng filter để thu hẹp phạm vi

# ❌ Quá rộng
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID

# ✅ Thu hẹp hơn
gcloud asset search-all-resources \
  --scope=organizations/ORG_ID \
  --asset-types=compute.googleapis.com/Instance \
  --query="location:us-central1"

Đổi trade-off giữa Export và Query

Dùng Real-time Query API khi:
- Cần trạng thái hiện tại
- Truy vấn một lần
- < 100k resources

Dùng BigQuery Exports khi:
- Cần phân tích lịch sử
- Truy vấn lặp lại (kinh tế hơn)
- Báo cáo tuân thủ quy mô lớn
- Phân tích xu hướng

Chi phí

Cloud Asset Inventory:
- Truy vấn thời gian thực: miễn phí (đã bao gồm)
- Export: $1 cho mỗi 1M resources mỗi tháng
- BigQuery: tính theo giá query chuẩn ($6.25 mỗi TB)

Ví dụ org với 100k resources:
- Export: $0.10/tháng
- Query: 10 queries × $0.01 mỗi query (ước tính) = $0.10/tháng
- Tổng: khoảng $0.20/tháng (rất rẻ)

Giám sát qua Policy Analyzer

Policy Analyzer phát hiện cấu hình IAM sai:

def analyze_iam_policy(org_id):
    """Kiểm tra IAM policies quá rộng"""
    
    from google.cloud import asset_v1
    
    client = asset_v1.AssetServiceClient()
    
    # Phân tích IAM policy
    analysis = client.analyze_iam_policy(
        analysis_query=asset_v1.IamPolicyAnalysisQuery(
            scope=f"organizations/{org_id}",
            resource_selector=asset_v1.IamPolicyAnalysisQuery.ResourceSelector(
                # Query tất cả resources
            ),
            access_selector=asset_v1.IamPolicyAnalysisQuery.AccessSelector(
                permissions=["*"]  # Tất cả permissions
            ),
            identity_selector=asset_v1.IamPolicyAnalysisQuery.IdentitySelector(
                identity="principalSet://goog/public:all"  # Public access
            )
        )
    )
    
    # Xem các finding về public access rủi ro
    for result in analysis.analysis_results:
        if result.fully_explored:
            # Resource cho phép public access
            print(f"⚠️  {result.resource} cho phép public access")

Mẫu tự động hóa

# Kiểm tra tuân thủ liên tục

def schedule_compliance_audit():
    """Chạy kiểm tra tuân thủ theo lịch"""
    
    from google.cloud import scheduler_v1
    
    # Tạo Cloud Scheduler job
    scheduler_client = scheduler_v1.CloudSchedulerClient()
    
    parent = f"projects/PROJECT_ID/locations/us-central1"
    
    job = scheduler_v1.Job(
        name=f"{parent}/jobs/compliance-audit",
        description="Daily compliance audit",
        schedule="0 2 * * *",  # 2 AM hằng ngày
        time_zone="UTC",
        http_target=scheduler_v1.HttpTarget(
            uri="https://us-central1-PROJECT_ID.cloudfunctions.net/audit",
            http_method=scheduler_v1.HttpMethod.POST
        )
    )
    
    scheduler_client.create_job(parent=parent, job=job)

# Cloud Function cho audit tuân thủ
def compliance_audit(request):
    """Cloud Function được scheduler kích hoạt"""
    
    findings = []
    findings.extend(audit_external_ips(ORG_ID))
    findings.extend(audit_owner_role(ORG_ID))
    
    if findings:
        notify_security_team(findings)
    
    return {"status": "completed", "findings": len(findings)}

Tham khảo

• Cloud Asset Inventory
• Search All Resources
• IAM Policy Analysis
• Security Command Center