防御OSS Bucket泄露：RAM权限策略+日志审计+敏感数据扫描三重防护

// 危险配置示例（通配符滥用）
{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "oss:*",
      "Resource": "*"  // 致命错误！
    }
  ]
}

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "oss:GetObject",
      "Resource": "acs:oss:cn-hangzhou:123456:bucket-finance/*",
      "Condition": {
        "IpAddress": {"acs:SourceIp": ["192.168.1.0/24"]}
      }
    }
  ]
}

# 启用Bucket Policy继承保护
aliyun oss bucket-policy --bucket-name mybucket \ 
    --policy '{
      "Statement":[
        {
          "Effect":"Deny",
          "Principal":"*",
          "Action":"*",
          "Resource":"acs:oss:*:*:mybucket/*",
          "Condition":{
            "StringNotLike":{"acs:Referer":["https://company.com/*"]}
          }
        }
      ]
    }'

# OSS访问日志示例
Time,SourceIP,Operation,Bucket,Object,HTTPStatus
2024-06-24T03:45:12Z,203.0.113.12,GetObject,finance-bucket,invoice.pdf,200
2024-06-24T03:46:51Z,198.51.100.78,PutObject,hr-bucket,salary.xlsx,403  # 异常点!

# 基于SLS的异常检测规则
def detect_anomaly(event):
    if event['Operation'] in ['DeleteObject', 'PutObjectAcl']:
        if event['SourceIP'] not in whitelist:
            send_alert(f"高危操作告警: {event['Operation']} by {event['SourceIP']}")
    
    if event['HTTPStatus'] == 403 and event['Bucket'] == 'finance-bucket':
        analyze_brute_force(event['SourceIP'])  # 暴力破解检测

# 敏感数据识别规则
- name: 身份证检测
  patterns: 
    - \b[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]\b
  risk_level: CRITICAL

- name: 银行卡识别
  patterns:
    - \b[1-9]\d{9,18}\b
  context_check: 
    keywords: ["卡号", "bank", "account"]

# 基于内容嗅探的优化逻辑
def need_scan(obj):
    # 跳过已扫描文件（ETag验证）
    if obj.etag in scanned_cache: 
        return False  
    
    # 根据扩展名过滤
    if obj.key.endswith(('.jpg','.mp4')): 
        return False
    
    # 小文件直接扫描，大文件抽样
    if obj.size > 100*1024*1024: 
        return random_sample(0.1)  # 10%抽样
    
    return True

# Terraform集成部署
module "oss_protection" {
  source = "git::https://protection-module"

  bucket_name     = "finance-data"
  ram_policy      = file("policies/finance_rw.json")
  scan_schedule   = "0 2 * * *"  # 每天2AM执行
  alert_webhook   = var.slack_webhook
}

# 防护系统日志输出
[RAM BLOCK] 2024-06-24T08:12:34Z IP:203.0.113.12 DENY GetObject 
  finance-bucket/confidential/employee_list.xlsx 
  Reason: IP not in whitelist

[SCAN ALERT] 2024-06-24T02:30:21Z Object: salary_template.docx 
  RiskType: ID_CARD Exposure Score: 92

# 生成带策略的临时Token
def gen_temp_token(user):
    policy = {
        "Version": "1",
        "Statement": [{
            "Effect": "Allow",
            "Action": "oss:GetObject",
            "Resource": f"acs:oss:*:*:{user.bucket}/{user.department}/*",
            "Condition": {"IpAddress": {"acs:SourceIp": user.ip}}
        }]
    }
    return aliyun.sts.assume_role(policy, duration=900)  # 15分钟有效期

-- 通过DataWorks实现查询脱敏
CREATE VIEW masked_employee AS
SELECT 
  id,
  name,
  mask(id_card) AS id_card,  -- 脱敏函数
  department 
FROM raw_employee_data;