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

# 建立簡單的密鑰
aws secretsmanager create-secret \
    --name MyDatabaseSecret \
    --description "My database credentials" \
    --secret-string '{"username":"admin","password":"MySecurePassword123!"}'

# 建立包含多個鍵值對的密鑰
aws secretsmanager create-secret \
    --name MyAPIKeys \
    --description "API keys for external services" \
    --secret-string '{
        "api_key": "abc123def456",
        "api_secret": "xyz789ghi012",
        "endpoint": "https://api.example.com"
    }'

 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

import boto3
import json

def create_secret(secret_name, secret_value, description=""):
    """建立新的密鑰"""
    client = boto3.client('secretsmanager')

    try:
        response = client.create_secret(
            Name=secret_name,
            Description=description,
            SecretString=json.dumps(secret_value)
        )
        print(f"密鑰建立成功: {response['ARN']}")
        return response
    except client.exceptions.ResourceExistsException:
        print(f"密鑰 {secret_name} 已存在")
        return None

# 使用範例
secret_data = {
    "username": "db_admin",
    "password": "SuperSecretPassword!",
    "host": "mydb.cluster-xxx.us-east-1.rds.amazonaws.com",
    "port": 5432,
    "dbname": "production"
}

create_secret(
    "prod/database/postgres",
    secret_data,
    "Production PostgreSQL database credentials"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 擷取密鑰值
aws secretsmanager get-secret-value \
    --secret-id MyDatabaseSecret \
    --query SecretString \
    --output text

# 擷取特定版本的密鑰
aws secretsmanager get-secret-value \
    --secret-id MyDatabaseSecret \
    --version-stage AWSPREVIOUS

 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

import boto3
import json
from botocore.exceptions import ClientError

def get_secret(secret_name, region_name="ap-northeast-1"):
    """擷取密鑰值"""
    client = boto3.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        response = client.get_secret_value(SecretId=secret_name)

        if 'SecretString' in response:
            secret = json.loads(response['SecretString'])
            return secret
        else:
            # 處理二進位密鑰
            import base64
            return base64.b64decode(response['SecretBinary'])

    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == 'DecryptionFailure':
            print("無法解密密鑰")
        elif error_code == 'InternalServiceError':
            print("內部服務錯誤")
        elif error_code == 'InvalidParameterException':
            print("無效參數")
        elif error_code == 'InvalidRequestException':
            print("無效請求")
        elif error_code == 'ResourceNotFoundException':
            print(f"找不到密鑰: {secret_name}")
        raise e

# 使用範例
credentials = get_secret("prod/database/postgres")
print(f"用戶名: {credentials['username']}")
print(f"主機: {credentials['host']}")

 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

const {
    SecretsManagerClient,
    GetSecretValueCommand
} = require("@aws-sdk/client-secrets-manager");

async function getSecret(secretName, region = "ap-northeast-1") {
    const client = new SecretsManagerClient({ region });

    try {
        const command = new GetSecretValueCommand({ SecretId: secretName });
        const response = await client.send(command);

        if (response.SecretString) {
            return JSON.parse(response.SecretString);
        }

        // 處理二進位密鑰
        const buffer = Buffer.from(response.SecretBinary, "base64");
        return buffer.toString("utf-8");

    } catch (error) {
        console.error(`擷取密鑰失敗: ${error.message}`);
        throw error;
    }
}

// 使用範例
(async () => {
    const credentials = await getSecret("prod/database/postgres");
    console.log(`用戶名: ${credentials.username}`);
    console.log(`主機: ${credentials.host}`);
})();

1
2
3
4
5

# 啟用自動輪換（使用預設 Lambda 函數）
aws secretsmanager rotate-secret \
    --secret-id MyRDSSecret \
    --rotation-lambda-arn arn:aws:lambda:ap-northeast-1:123456789012:function:SecretsManagerRDSPostgreSQLRotation \
    --rotation-rules "{\"AutomaticallyAfterDays\": 30}"

 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

AWSTemplateFormatVersion: '2010-09-09'
Description: 'Secrets Manager with rotation'

Resources:
  MyDatabaseSecret:
    Type: AWS::SecretsManager::Secret
    Properties:
      Name: prod/database/credentials
      Description: Production database credentials
      GenerateSecretString:
        SecretStringTemplate: '{"username": "admin"}'
        GenerateStringKey: password
        PasswordLength: 32
        ExcludeCharacters: '"@/\'

  SecretRotationSchedule:
    Type: AWS::SecretsManager::RotationSchedule
    Properties:
      SecretId: !Ref MyDatabaseSecret
      RotationLambdaARN: !GetAtt RotationLambda.Arn
      RotationRules:
        AutomaticallyAfterDays: 30

  RotationLambda:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: secrets-rotation-function
      Runtime: python3.9
      Handler: lambda_function.lambda_handler
      Role: !GetAtt RotationLambdaRole.Arn
      Code:
        S3Bucket: my-lambda-bucket
        S3Key: rotation-function.zip
      Timeout: 30
      VpcConfig:
        SecurityGroupIds:
          - !Ref LambdaSecurityGroup
        SubnetIds:
          - !Ref PrivateSubnet1
          - !Ref PrivateSubnet2

  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
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107

import boto3
import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    """密鑰輪換 Lambda 處理程式"""
    arn = event['SecretId']
    token = event['ClientRequestToken']
    step = event['Step']

    client = boto3.client('secretsmanager')

    # 輪換包含四個步驟
    if step == "createSecret":
        create_secret(client, arn, token)
    elif step == "setSecret":
        set_secret(client, arn, token)
    elif step == "testSecret":
        test_secret(client, arn, token)
    elif step == "finishSecret":
        finish_secret(client, arn, token)
    else:
        raise ValueError(f"無效的步驟: {step}")

def create_secret(client, arn, token):
    """建立新的密鑰版本"""
    # 取得目前密鑰
    current_secret = client.get_secret_value(
        SecretId=arn,
        VersionStage="AWSCURRENT"
    )

    # 產生新密碼
    new_password = client.get_random_password(
        PasswordLength=32,
        ExcludeCharacters='"@/\\'
    )['RandomPassword']

    # 更新密鑰值
    secret_dict = json.loads(current_secret['SecretString'])
    secret_dict['password'] = new_password

    # 儲存為待處理版本
    client.put_secret_value(
        SecretId=arn,
        ClientRequestToken=token,
        SecretString=json.dumps(secret_dict),
        VersionStages=['AWSPENDING']
    )
    logger.info(f"已建立新的密鑰版本: {token}")

def set_secret(client, arn, token):
    """在資料庫中設定新密碼"""
    # 取得待處理密鑰
    pending = client.get_secret_value(
        SecretId=arn,
        VersionId=token,
        VersionStage="AWSPENDING"
    )

    secret_dict = json.loads(pending['SecretString'])

    # 連接資料庫並更新密碼
    # 這裡需要根據您的資料庫類型實作
    update_database_password(secret_dict)

    logger.info("資料庫密碼已更新")

def test_secret(client, arn, token):
    """測試新密鑰是否可用"""
    pending = client.get_secret_value(
        SecretId=arn,
        VersionId=token,
        VersionStage="AWSPENDING"
    )

    secret_dict = json.loads(pending['SecretString'])

    # 嘗試使用新憑證連接資料庫
    if not test_database_connection(secret_dict):
        raise ValueError("新密鑰測試失敗")

    logger.info("新密鑰測試成功")

def finish_secret(client, arn, token):
    """完成輪換，將待處理版本設為目前版本"""
    # 取得目前版本 ID
    metadata = client.describe_secret(SecretId=arn)
    current_version = None

    for version_id, stages in metadata['VersionIdsToStages'].items():
        if 'AWSCURRENT' in stages:
            current_version = version_id
            break

    # 更新版本階段
    client.update_secret_version_stage(
        SecretId=arn,
        VersionStage='AWSCURRENT',
        MoveToVersionId=token,
        RemoveFromVersionId=current_version
    )

    logger.info(f"密鑰輪換完成: {token}")

1
2
3
4
5

// build.gradle
dependencies {
    implementation 'software.amazon.awssdk:secretsmanager:2.20.0'
    implementation 'com.google.code.gson:gson:2.10.1'
}

 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
45
46

// SecretsManagerService.java
package com.example.service;

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.secretsmanager.SecretsManagerClient;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueRequest;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueResponse;
import com.google.gson.Gson;
import org.springframework.stereotype.Service;

import java.util.Map;

@Service
public class SecretsManagerService {

    private final SecretsManagerClient client;
    private final Gson gson;

    public SecretsManagerService() {
        this.client = SecretsManagerClient.builder()
            .region(Region.AP_NORTHEAST_1)
            .build();
        this.gson = new Gson();
    }

    public Map<String, String> getSecret(String secretName) {
        GetSecretValueRequest request = GetSecretValueRequest.builder()
            .secretId(secretName)
            .build();

        GetSecretValueResponse response = client.getSecretValue(request);

        return gson.fromJson(response.secretString(), Map.class);
    }

    public String getDatabaseUrl(String secretName) {
        Map<String, String> secret = getSecret(secretName);

        return String.format(
            "jdbc:postgresql://%s:%s/%s",
            secret.get("host"),
            secret.get("port"),
            secret.get("dbname")
        );
    }
}

 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

# external-secret.yaml
apiVersion: external-secrets.io/v1beta1
kind: ExternalSecret
metadata:
  name: database-credentials
  namespace: production
spec:
  refreshInterval: 1h
  secretStoreRef:
    name: aws-secrets-manager
    kind: ClusterSecretStore
  target:
    name: database-credentials
    creationPolicy: Owner
  data:
    - secretKey: username
      remoteRef:
        key: prod/database/postgres
        property: username
    - secretKey: password
      remoteRef:
        key: prod/database/postgres
        property: password
    - secretKey: host
      remoteRef:
        key: prod/database/postgres
        property: host

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

# cluster-secret-store.yaml
apiVersion: external-secrets.io/v1beta1
kind: ClusterSecretStore
metadata:
  name: aws-secrets-manager
spec:
  provider:
    aws:
      service: SecretsManager
      region: ap-northeast-1
      auth:
        jwt:
          serviceAccountRef:
            name: external-secrets
            namespace: external-secrets

 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
45
46
47
48
49
50
51
52
53
54
55

import boto3
import json
from datetime import datetime, timedelta
from threading import Lock

class SecretsCache:
    """帶有快取功能的密鑰管理器"""

    def __init__(self, cache_ttl_seconds=300):
        self.client = boto3.client('secretsmanager')
        self.cache = {}
        self.cache_ttl = timedelta(seconds=cache_ttl_seconds)
        self.lock = Lock()

    def get_secret(self, secret_name):
        """擷取密鑰（優先從快取）"""
        with self.lock:
            # 檢查快取
            if secret_name in self.cache:
                cached_item = self.cache[secret_name]
                if datetime.now() < cached_item['expires_at']:
                    return cached_item['value']

            # 從 Secrets Manager 擷取
            response = self.client.get_secret_value(SecretId=secret_name)
            secret_value = json.loads(response['SecretString'])

            # 更新快取
            self.cache[secret_name] = {
                'value': secret_value,
                'expires_at': datetime.now() + self.cache_ttl
            }

            return secret_value

    def invalidate(self, secret_name=None):
        """使快取失效"""
        with self.lock:
            if secret_name:
                self.cache.pop(secret_name, None)
            else:
                self.cache.clear()

# 全域快取實例
secrets_cache = SecretsCache(cache_ttl_seconds=300)

# 使用範例
def get_database_connection():
    credentials = secrets_cache.get_secret("prod/database/postgres")
    return create_connection(
        host=credentials['host'],
        user=credentials['username'],
        password=credentials['password'],
        database=credentials['dbname']
    )