I manage a fleet of IoT devices.
As for now, GitHub announced that the username-password authentication would be deprecative soon, so I have to change the password on each device to the GitHub access token. I store the new token at AWS secrets manager. And those are the scripts to extract the new token and implement it on the device.
update_github_token.py
from get_aws_secret_manager import get_secret
import subprocess
SECRET_NAME = 'prod/GitHub/Token'
GITHUB_TOKEN_KEY = 'droneGitHubToken'
GITHUB_USER_NAME = 'danAairlines'
GITHUB_REPO_OWNER = 'Aairlinesfox'
def get_github_tokken():
github_secret_dict = get_secret(SECRET_NAME)
return github_secret_dict[GITHUB_TOKEN_KEY]
def build_token_github_url():
github_token = get_github_tokken()
return f'https://{GITHUB_USER_NAME}:{github_token}@github.com/{GITHUB_REPO_OWNER}/drone.git'
def update_drone_github_url():
github_url = build_token_github_url()
subprocess.run(f'sudo git remote set-url origin {github_url}', shell=True, check=True)
if __name__ == '__main__':
update_drone_github_url()
The second module is AWS script, as appears on their website. I just modified the get_secret function by adding args. And at the end of the file, I added:
secret = json.loads(get_secret_value_response['SecretString'])
return secret
get_aws_secret_manager.py
# If you need more information about configurations or implementing the sample code, visit the AWS docs:
# https://aws.amazon.com/developers/getting-started/python/
import boto3
import base64
from botocore.exceptions import ClientError
import json
def get_secret(aws_secret_name: str, aws_region_name: str = "eu-central-1"):
secret_name = aws_secret_name
region_name = aws_region_name
# Create a Secrets Manager client
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
# 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 = json.loads(get_secret_value_response['SecretString'])
return secret
else:
decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])```