Rotirea cheilor AWS IAM Access folosind funcția Lambda
Conform celor mai bune practici de securitate AWS, ar trebui să ne rotim în mod regulat cheile de acces IAM pentru a ne îmbunătăți securitatea conturilor AWS.
În acest articol vă voi învăța cumautomatizaprocesul de rotație a cheilor de acces IAM în funcție de vechimea cheii de acces folosind funcția Lambda.
Use this link to rotate the keys manually using AWS document https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html#Using_RotateAccessKey
AWS IAM (Managementul identității și accesului) oferă control al accesului detaliat pe tot AWS.
Chei AWS Accesssunt acreditări pe termen lung pentru un utilizator IAM. Puteți utiliza cheile de acces pentru a semna solicitări programatice la AWS CLI sau AWS API (direct sau utilizând SDK-ul AWS).
AWS Lambdaeste un serviciu de calcul fără server, bazat pe evenimente, care vă permite să rulați cod pentru aproape orice tip de aplicație sau serviciu backend fără a furniza sau gestiona servere. Puteți declanșa Lambda din peste 200 de servicii AWS și software ca serviciu (SaaS) și plătiți doar pentru ceea ce utilizați.
Diagrama arhitecturii
Rotirea cheilor de acces IAM utilizând funcția AWS Lambda și trimiterea notificărilor utilizând AWS SES
Diagrama arată următorul flux de lucru:
1- Utilizatorii IAM se vor conecta programatic folosind cheile de acces IAM
2- Evenimentul CloudWatch inițiază o funcție Lambda la fiecare 24 de ore
3- Funcția Lambda inițiază o funcție Lambda pentru fiecare ID de cont AWS și îi transmite metadatele pentru procesare suplimentară. Acesta va verifica vârstele cheie de acces la toți utilizatorii și va iniția e-mailul către proprietarul contului
4- În cazul meu, am configurat cele trei condiții pentru a trimite e-mailuri folosind AWS SES.
Script Python în funcția Lambda:
Am scris scriptul de rotație a tastelor de acces în python. Acest script python va obține mai întâi informațiile/metadatele fiecărui utilizator de la AWS IAM. După cum știți, utilizatorul AWS IAM ar putea fi om sau ar putea fi mașini, așa că am etichetat corect fiecare utilizator cu următoarele informații:
- Nume
- Tip de utilizator (angajat sau mașină)
- E-mail (Abc@xyz.com)
Recomandat de LinkedIn
Funcția Lambda va obține aceste informații de mai sus de la etichetele de utilizator IAM și dacă UserType este "angajat" și cheia are 70, 80 sau 90 de zile, apoi va trimite e-mailul de notificare folosind AWS SES și va dezactiva cheia de acces. Dacă UserType este "mașină" și vârsta cheie 70, 80 sau 90 de zile, atunci va trimite e-mailul de notificare echipei în cauză (cum ar fi IT, Ops etc.).
Trimiteți alerte și dezactivați condițiile cheilor de acces:
Prima condiție dacă vârsta cheii de acces IAM a utilizatorului este a 70-a zi, apoi trimiteți e-mailul de alertă"Astăzi este a 70-a zi a cheilor de acces. Vă rugăm să rotiți cheia de acces înainte de a împlini 90 de zile".
A doua condiție: dacă vârsta cheii de acces IAM a utilizatorului este a 80-a zi, atunci trimiteți e-mailul de alertă"Astăzi este a 80-a zi a cheilor de acces. Vă rugăm să rotiți cheia de acces înainte de a împlini 90 de zile".
A treia condiție: dacă vârsta cheii de acces IAM a utilizatorului este a 90-a zi, atunci trimiteți e-mailul de alertă"Astăzi este cea de-a 90-a zi de la cheile de acces și cheia ta a fost inactivată. Vă rugăm să generați o nouă cheie sau să contactați echipa de asistență IT".
Codul funcției Lambda:
Obțineți codul formatat corespunzător din depozitul meu github: https://github.com/engr-usman/blogs.git
import boto3, os, time, datetime, sys, jso
from datetime import date
from botocore.exceptions import ClientError
iam = boto3.client('iam')
def lambda_handler(event, context):
email_70_list = []
email_80_list = []
email_90_list = []
# print("All IAM user emails that have AccessKeys 70 days or older")
unique_user_list = (iam.list_users()['Users'])
for userlist in unique_user_list:
userKeys = iam.list_access_keys(UserName=userlist['UserName'])
for keyValue in userKeys['AccessKeyMetadata']:
UserAccessKeyID = keyValue['AccessKeyId']
IAMUserName = keyValue['UserName']
#print(f"IAMUserName IAM Users:{len(IAMUserName)}: {IAMUserName}")
if keyValue['Status'] == 'Active':
currentdate = date.today()
active_days = currentdate - keyValue['CreateDate'].date()
#print ("The active days details are: ", active_days)
#print ("datetime details are: ", datetime.timedelta(days=15))
# if Access key age is greater then or equal to 70 days, send warning
if active_days == datetime.timedelta(days=int(os.environ['days_70'])):
userTags = iam.list_user_tags(UserName=keyValue['UserName'])
email_tag = list(filter(lambda tag: tag['Key'] == 'email', userTags['Tags']))
if(len(email_tag) == 1):
email = email_tag[0]['Value']
email_70_list.append(email)
print("This User: ", IAMUserName, ", with the email: ", email, ", is having access key age is 70 days")
email_unique = list(set(email_70_list))
print("Email list: ", email_unique)
RECIPIENTS = email_unique
SENDER = os.environ['sender_email']
AWS_REGION = os.environ['region']
SUBJECT_70 = os.environ['SUBJECT_70']
BODY_TEXT_70 = os.environ['BODY_TEXT_70']
BODY_HTML_70 = os.environ['BODY_HTML_70']
CHARSET = "UTF-8"
client = boto3.client('ses',region_name=AWS_REGION)
try:
response = client.send_email(
Destination={
'ToAddresses': RECIPIENTS,
},
Message={
'Body': {
'Html': {
'Charset': CHARSET,
'Data': BODY_HTML_70,
},
'Text': {
'Charset': CHARSET,
'Data': BODY_TEXT_70,
},
},
'Subject': {
'Charset': CHARSET,
'Data': SUBJECT_70,
},
},
Source=SENDER,
)
except ClientError as e:
print(e.response['Error']['Message'])
else:
print("Email sent! Message ID:"),
print(response['MessageId'])
# if Access Key Age is greater then 80 days, send email alert
if active_days == datetime.timedelta(days=int(os.environ['days_80'])):
userTags = iam.list_user_tags(UserName=keyValue['UserName'])
email_tag = list(filter(lambda tag: tag['Key'] == 'email', userTags['Tags']))
if(len(email_tag) == 1):
email = email_tag[0]['Value']
email_80_list.append(email)
print("The User: ", IAMUserName, ", with the email: ", email, ", is having access key age is 80 days")
email_unique = list(set(email_80_list))
print("Email list: ", email_unique)
RECIPIENTS = email_unique
SENDER = os.environ['sender_email']
print("Sender: ", SENDER)
AWS_REGION = os.environ['region']
SUBJECT_80 = os.environ['SUBJECT_80']
BODY_TEXT_80 = os.environ['BODY_TEXT_80']
BODY_HTML_80 = os.environ['BODY_HTML_80']
CHARSET = "UTF-8"
client = boto3.client('ses',region_name=AWS_REGION)
try:
response = client.send_email(
Destination={
'ToAddresses': RECIPIENTS,
},
Message={
'Body': {
'Html': {
'Charset': CHARSET,
'Data': BODY_HTML_80,
},
'Text': {
'Charset': CHARSET,
'Data': BODY_TEXT_80,
},
},
'Subject': {
'Charset': CHARSET,
'Data': SUBJECT_80,
},
},
Source=SENDER,
)
except ClientError as e:
print(e.response['Error']['Message'])
else:
print("Email sent! Message ID:"),
print(response['MessageId'])
# if Access Key Age is greater then 90 days, send email alert and inactive access keys
if active_days >= datetime.timedelta(days=int(os.environ['days_90'])):
userTags = iam.list_user_tags(UserName=keyValue['UserName'])
email_tag = list(filter(lambda tag: tag['Key'] == 'email', userTags['Tags']))
user1_tag = list(filter(lambda tag: tag['Key'] == 'UserType', userTags['Tags']))
if(len(email_tag) == 1):
email = email_tag[0]['Value']
email_90_list.append(email)
print("The User: ", IAMUserName, ", with the email: ", email, ", is having access key age is greater then 90 days")
if(len(user1_tag) == 1):
user1tag = user1_tag[0]['Value']
if user1tag == "Employee":
iam.update_access_key(AccessKeyId=UserAccessKeyID,Status='Inactive',UserName=IAMUserName)
print("Status has been updated to Inactive")
email_unique = list(set(email_90_list))
print("Email list: ", email_unique)
RECIPIENTS = email_unique
SENDER = os.environ['sender_email']
print("Sender: ", SENDER)
AWS_REGION = os.environ['region']
SUBJECT_90 = os.environ['SUBJECT_90']
BODY_TEXT_90 = os.environ['BODY_TEXT_90']
BODY_HTML_90 = os.environ['BODY_HTML_90']
CHARSET = "UTF-8"
client = boto3.client('ses',region_name=AWS_REGION)
try:
response = client.send_email(
Destination={
'ToAddresses': RECIPIENTS,
},
Message={
'Body': {
'Html': {
'Charset': CHARSET,
'Data': BODY_HTML_90,
},
'Text': {
'Charset': CHARSET,
'Data': BODY_TEXT_90,
},
},
'Subject': {
'Charset': CHARSET,
'Data': SUBJECT_90,
},
},
Source=SENDER,
)
except ClientError as e:
print(e.response['Error']['Message'])
else:
print("Email sent! Message ID:"),
print(response['MessageId'])n
Notă: Indentarea nu este corectă a codului de mai sus:( și știu că există o mulțime de îmbunătățiri ale codului, este doar pentru o idee.
Activați regula de eveniment CloudWatch:
Acum trebuie doar să creăm o regulă de eveniment CloudWatch (acum AWS EventBridge) pentru a declanșa această funcție Lambda la o anumită oră. În cazul meu, am activat acest job cron la 9 dimineața.
Programator și țintă AWS EventBridge
Concluzie
Aceasta este cea mai bună practică de securitate AWS pentru a roti cheile de acces IAM, astfel încât nimeni să nu poată accesa resursele contului dvs. Funcția Lambda este cel mai bun instrument pentru a efectua acest tip de sarcini și a le declanșa de la alte servicii precum CloudWatch.
Sper că v-a plăcut să citiți acest articol, vă rog să nu ezitați să mă contactați dacă aveți întrebări.
Usman Ahmad what are the roles you provided your lambda function ?
Hi Usman Ahmad I am trying to fetch the details of both sender and region details here. bot its not happening. pls let me know whats this sender refers to? how i should try in my code. as you said in the post i have created the mail in ses and mentioned the VAULE in the IAM user tags. can you help me what i am missing inorder to get to get the region and sender details
Hi Usman! Can you publish script in the correct format (at least screenshot), please? I'm not Python specialist for now, maybe in the future. Thank you.
Very useful
Amazing