云安全与隐私保护

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

{
  "Version": "2012-10-17",
  "Statement": [{
    "Sid": "ReadOnlyFromOfficeWithMfa",
    "Effect": "Allow",
    "Action": ["s3:GetObject", "s3:ListBucket"],
    "Resource": [
      "arn:aws:s3:::reports-prod",
      "arn:aws:s3:::reports-prod/*"
    ],
    "Condition": {
      "IpAddress":  { "aws:SourceIp": "203.0.113.0/24" },
      "Bool":       { "aws:MultiFactorAuthPresent": "true" },
      "NumericLessThan": { "aws:MultiFactorAuthAge": "3600" }
    }
  }]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DenyOutsideApprovedRegions",
      "Effect": "Deny",
      "Action": "*",
      "Resource": "*",
      "Condition": {
        "StringNotEquals": {
          "aws:RequestedRegion": ["us-east-1", "us-west-2", "eu-west-1"]
        }
      }
    },
    {
      "Sid": "DenyRootUser",
      "Effect": "Deny",
      "Action": "*",
      "Resource": "*",
      "Condition": {
        "StringEquals": { "aws:PrincipalType": "Root" }
      }
    }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import boto3

kms = boto3.client("kms")

# 信封加密：KMS 加密一个小的数据密钥；
# 你在本地用这个数据密钥加密真正的负载。
def encrypt_blob(plaintext: bytes, key_id: str) -> dict:
    resp = kms.generate_data_key(KeyId=key_id, KeySpec="AES_256")
    plaintext_dk, encrypted_dk = resp["Plaintext"], resp["CiphertextBlob"]

    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    import os
    nonce = os.urandom(12)
    ciphertext = AESGCM(plaintext_dk).encrypt(nonce, plaintext, associated_data=None)

    # 丢弃明文数据密钥；只持久化加密后的版本。
    return {"nonce": nonce, "ciphertext": ciphertext, "wrapped_key": encrypted_dk}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

server {
    listen 443 ssl http2;
    listen [::]:443 ssl http2;

    ssl_protocols       TLSv1.2 TLSv1.3;
    ssl_ciphers         ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-CHACHA20-POLY1305;
    ssl_prefer_server_ciphers on;

    ssl_session_cache   shared:SSL:10m;
    ssl_session_timeout 1h;
    ssl_session_tickets off;

    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;
    add_header X-Content-Type-Options    "nosniff"  always;
    add_header X-Frame-Options           "DENY"     always;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

{
  "Rules": [
    {
      "Name": "AWSManagedRulesCommonRuleSet",
      "Priority": 10,
      "OverrideAction": { "None": {} },
      "Statement": {
        "ManagedRuleGroupStatement": {
          "VendorName": "AWS",
          "Name": "AWSManagedRulesCommonRuleSet"
        }
      },
      "VisibilityConfig": { "SampledRequestsEnabled": true,
                            "CloudWatchMetricsEnabled": true,
                            "MetricName": "common" }
    },
    {
      "Name": "RateLimitPerIp",
      "Priority": 20,
      "Action": { "Block": {} },
      "Statement": {
        "RateBasedStatement": { "Limit": 2000, "AggregateKeyType": "IP" }
      },
      "VisibilityConfig": { "SampledRequestsEnabled": true,
                            "CloudWatchMetricsEnabled": true,
                            "MetricName": "ratelimit" }
    }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
  "eventTime": "2024-01-15T14:30:00Z",
  "eventSource": "s3.amazonaws.com",
  "eventName": "GetObject",
  "userIdentity": {
    "type": "AssumedRole",
    "arn": "arn:aws:sts::111122223333:assumed-role/AnalystRole/Alice",
    "sessionContext": { "mfaAuthenticated": "true" }
  },
  "sourceIPAddress": "203.0.113.12",
  "userAgent": "aws-sdk-python/1.28",
  "requestParameters": { "bucketName": "reports-prod", "key": "q4-revenue.csv" },
  "additionalEventData": { "bytesTransferredOut": 38421 }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

"""被入侵的 EC2 实例：取证快照、隔离、停机。"""
import boto3
from datetime import datetime

ec2 = boto3.client("ec2")

def quarantine(instance_id: str, reason: str) -> dict:
    timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    tag = [{"Key": "incident",  "Value": reason},
           {"Key": "timestamp", "Value": timestamp}]

    # 1. 在做任何其他事情之前，先给每一块挂载卷打快照。
    inst = ec2.describe_instances(InstanceIds=[instance_id])["Reservations"][0]["Instances"][0]
    snapshots = []
    for bdm in inst["BlockDeviceMappings"]:
        snap = ec2.create_snapshot(
            VolumeId=bdm["Ebs"]["VolumeId"],
            Description=f"forensic-{instance_id}-{reason}-{timestamp}",
            TagSpecifications=[{"ResourceType": "snapshot", "Tags": tag}],
        )
        snapshots.append(snap["SnapshotId"])

    # 2. 把安全组替换成"什么都不放行"。
    isolation_sg = ec2.create_security_group(
        GroupName=f"quarantine-{instance_id}-{timestamp}",
        Description=f"Incident isolation: {reason}",
        VpcId=inst["VpcId"],
    )["GroupId"]
    ec2.modify_instance_attribute(InstanceId=instance_id, Groups=[isolation_sg])

    # 3. 解绑 IAM 实例 Profile，让凭证无法继续被复用。
    profile = inst.get("IamInstanceProfile", {}).get("Arn")
    if profile:
        association_id = next(
            a["AssociationId"]
            for a in ec2.describe_iam_instance_profile_associations(
                Filters=[{"Name": "instance-id", "Values": [instance_id]}])["IamInstanceProfileAssociations"]
        )
        ec2.disassociate_iam_instance_profile(AssociationId=association_id)

    # 4. 停机（不要 terminate——terminate 会销毁证据）。
    ec2.stop_instances(InstanceIds=[instance_id])

    return {"instance": instance_id, "snapshots": snapshots, "isolation_sg": isolation_sg}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

resource "aws_s3_bucket" "data" {
  bucket = "reports-${var.environment}"

  tags = {
    DataClassification = "confidential"
    Owner              = "analytics"
    Compliance         = "soc2,hipaa"
  }
}

resource "aws_s3_bucket_versioning" "data" {
  bucket = aws_s3_bucket.data.id
  versioning_configuration { status = "Enabled" }
}

resource "aws_s3_bucket_server_side_encryption_configuration" "data" {
  bucket = aws_s3_bucket.data.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = aws_kms_key.reports.arn
    }
    bucket_key_enabled = true   # 把 KMS 调用费用降约 99%
  }
}

resource "aws_s3_bucket_public_access_block" "data" {
  bucket                  = aws_s3_bucket.data.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "aws_s3_bucket_logging" "data" {
  bucket        = aws_s3_bucket.data.id
  target_bucket = aws_s3_bucket.access_logs.id
  target_prefix = "reports/"
}