如何通过 python 更新 AWS Secrets Manager?
How to update AWS Secrets Manager via python?
我找不到任何关于如何 upload/update 向 AWS 机密管理器赋值的文档。我只能通过 python 检索值。有解决办法吗?
您可以使用 update_secret()
:
response = client.update_secret(
SecretId='string',
ClientRequestToken='string',
Description='string',
KmsKeyId='string',
SecretBinary=b'bytes',
SecretString='string'
)
要创建新 机密,请使用:put_secret_value()
import json
from boto3 import Session
# initialize session client
session = Session(
aws_access_key_id="aws_access_key_id",
aws_secret_access_key="aws_secret_access_key",
region_name="region_name"
)
client = session.client(service_name="secretsmanager")
FOR CREATE
client.create_secret(Name="my_first_secret", SecretString=json.dumps({"favorite_character": "stitch!"}))
FOR UPDATE
# get original secrets
original_secret = client.get_secret_value(SecretId="my_first_secret")
# update secrets
updated_secret = original_secret.update({"UPDATE_KEY": "update_value"})
client.update_secret(SecretId="my_secret_name", SecretString=json.dumps(updated_secret))
def init_aws_session():
region_name = "us-east-1"
my_access_id = 'my_access_id'
my_secret_key = 'my_secret_key'
# Create a Secrets Manager client
session = boto3.session.Session(
region_name=region_name,
aws_access_key_id=my_access_id,
aws_secret_access_key=my_secret_key
)
client = session.client(
service_name='secretsmanager',
region_name=region_name,
)
return client
def update_secret(secret_name, key, value):
client = init_aws_session()
# get original secrets
config_secret = get_secret(secret_name, client)
secret.update({key: value})
client.update_secret(SecretId=secret_name, SecretString=json.dumps(secret))
print(secret)
def get_secret(secret_name):
client = init_aws_session()
# In this sample we only handle the specific exceptions for the 'GetSecretValue' API.
# See https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
# We rethrow the exception by default.
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
if e.response['Error']['Code'] == 'DecryptionFailureException':
# Secrets Manager can't decrypt the protected secret text using the provided KMS key.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InternalServiceErrorException':
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidParameterException':
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidRequestException':
# You provided a parameter value that is not valid for the current state of the resource.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'ResourceNotFoundException':
# We can't find the resource that you asked for.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
else:
# Decrypts secret using the associated KMS CMK.
# Depending on whether the secret is a string or binary, one of these fields will be populated.
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
# Your code goes here.
return json.loads(secret)
if __name__ == '__main__':
update_secret(some_secret, key, value)
我找不到任何关于如何 upload/update 向 AWS 机密管理器赋值的文档。我只能通过 python 检索值。有解决办法吗?
您可以使用 update_secret()
:
response = client.update_secret(
SecretId='string',
ClientRequestToken='string',
Description='string',
KmsKeyId='string',
SecretBinary=b'bytes',
SecretString='string'
)
要创建新 机密,请使用:put_secret_value()
import json
from boto3 import Session
# initialize session client
session = Session(
aws_access_key_id="aws_access_key_id",
aws_secret_access_key="aws_secret_access_key",
region_name="region_name"
)
client = session.client(service_name="secretsmanager")
FOR CREATE
client.create_secret(Name="my_first_secret", SecretString=json.dumps({"favorite_character": "stitch!"}))
FOR UPDATE
# get original secrets
original_secret = client.get_secret_value(SecretId="my_first_secret")
# update secrets
updated_secret = original_secret.update({"UPDATE_KEY": "update_value"})
client.update_secret(SecretId="my_secret_name", SecretString=json.dumps(updated_secret))
def init_aws_session():
region_name = "us-east-1"
my_access_id = 'my_access_id'
my_secret_key = 'my_secret_key'
# Create a Secrets Manager client
session = boto3.session.Session(
region_name=region_name,
aws_access_key_id=my_access_id,
aws_secret_access_key=my_secret_key
)
client = session.client(
service_name='secretsmanager',
region_name=region_name,
)
return client
def update_secret(secret_name, key, value):
client = init_aws_session()
# get original secrets
config_secret = get_secret(secret_name, client)
secret.update({key: value})
client.update_secret(SecretId=secret_name, SecretString=json.dumps(secret))
print(secret)
def get_secret(secret_name):
client = init_aws_session()
# In this sample we only handle the specific exceptions for the 'GetSecretValue' API.
# See https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
# We rethrow the exception by default.
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
if e.response['Error']['Code'] == 'DecryptionFailureException':
# Secrets Manager can't decrypt the protected secret text using the provided KMS key.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InternalServiceErrorException':
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidParameterException':
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidRequestException':
# You provided a parameter value that is not valid for the current state of the resource.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'ResourceNotFoundException':
# We can't find the resource that you asked for.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
else:
# Decrypts secret using the associated KMS CMK.
# Depending on whether the secret is a string or binary, one of these fields will be populated.
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
# Your code goes here.
return json.loads(secret)
if __name__ == '__main__':
update_secret(some_secret, key, value)