GCP GCS Object Copied to Different Bucket
Description
Detects when GCS objects are copied from one bucket to a bucket in a different GCP project. Cross-project copies are more suspicious than same-project copies and can indicate data exfiltration where an adversary copies sensitive data to a project they control. The threshold of 50+ copy operations suggests bulk exfiltration rather than normal operations. This is detected by monitoring storage.objects.get operations that include a destination field in the metadata, indicating a copy operation.
Query · python
def _parse_destination(destination):
"""Parse bucket and project from destination path.
Returns tuple: (dest_bucket, dest_project)
destination format: projects/PROJECT_ID/buckets/BUCKET_NAME/objects/...
"""
dest_bucket = None
dest_project = None
if "buckets/" in destination and "objects/" in destination:
try:
dest_bucket = destination.split("buckets/")[1].split("/objects/")[0]
except (IndexError, AttributeError):
pass
if "projects/" in destination and "buckets/" in destination:
try:
project = destination.split("projects/")[1].split("/buckets/")[0]
dest_project = project if project != "_" else None
except (IndexError, AttributeError):
pass
return dest_bucket, dest_project
def rule(event):
if (
event.deep_get("protoPayload", "methodName") != "storage.objects.get"
or event.deep_get("protoPayload", "serviceName") != "storage.googleapis.com"
or event.get("severity") == "ERROR" # Operation failed
or not event.deep_get("protoPayload", "metadata", "destination")
):
return False
# Extract source and destination buckets and projects
source_bucket = event.deep_get("resource", "labels", "bucket_name")
source_project = event.deep_get("resource", "labels", "project_id")
destination = event.deep_get("protoPayload", "metadata", "destination", default="")
dest_bucket, dest_project = _parse_destination(destination)
# Validate required fields
if not all([source_bucket, dest_bucket, source_project, dest_project]):
return False
# Only alert on cross-project copies (more suspicious than same-project copies)
is_different_bucket = source_bucket != dest_bucket
is_cross_project = dest_project != source_project
return is_different_bucket and is_cross_project
def severity(event):
"""Dynamic severity based on whether destination is in a different project."""
source_project = event.deep_get("resource", "labels", "project_id")
destination = event.deep_get("protoPayload", "metadata", "destination", default="")
_, dest_project = _parse_destination(destination)
if dest_project and source_project and dest_project != source_project:
return "DEFAULT"
return "LOW"
def title(event):
source_bucket = event.deep_get("resource", "labels", "bucket_name", default="Unknown")
source_project = event.deep_get("resource", "labels", "project_id", default="Unknown")
destination = event.deep_get("protoPayload", "metadata", "destination", default="")
dest_bucket, dest_project = _parse_destination(destination)
if not dest_bucket:
dest_bucket = "Unknown"
actor = event.deep_get(
"protoPayload", "authenticationInfo", "principalEmail", default="Unknown"
)
# Different title based on cross-project vs same-project
if dest_project and source_project != "Unknown" and dest_project != source_project:
return (
f"CROSS-PROJECT: GCS object copied from "
f"[{source_project}/{source_bucket}] to "
f"[{dest_project}/{dest_bucket}] by [{actor}]"
)
return (
f"GCS object copied from bucket " f"[{source_bucket}] to [{dest_bucket}] " f"by [{actor}]"
)
def alert_context(event):
destination = event.deep_get("protoPayload", "metadata", "destination", default="")
dest_bucket, dest_project = _parse_destination(destination)
source_project = event.deep_get("resource", "labels", "project_id")
is_cross_project = dest_project and source_project and dest_project != source_project
return {
"principal": event.deep_get("protoPayload", "authenticationInfo", "principalEmail"),
"source_project": source_project,
"source_bucket": event.deep_get("resource", "labels", "bucket_name"),
"destination_project": dest_project,
"destination_bucket": dest_bucket,
"is_cross_project": is_cross_project,
"destination_path": destination,
"source_object": event.deep_get("protoPayload", "resourceName"),
"source_ip": event.deep_get("protoPayload", "requestMetadata", "callerIp"),
"user_agent": event.deep_get("protoPayload", "requestMetadata", "callerSuppliedUserAgent"),
"bytes_requested": event.deep_get("protoPayload", "metadata", "requested_bytes"),
}
Analyst notes
- Query GCP Audit logs for all storage.objects.get operations with destination metadata by authenticationInfo:principalEmail in the 2 hours around this alert to identify the full scope of copy operations
- Check if the protoPayload:metadata:destination bucket belongs to the same project, a different project in the organization, or an external attacker-controlled project
- Review GCP Audit logs for IAM permission changes, service account key creation, or bucket policy modifications by this principal in the 24 hours before the first copy operation