AWS SDK로 읽기 전용·읽기/쓰기 IAM 사용자 만들기
AWS SDK로 읽기 전용·읽기/쓰기 IAM 사용자 만들기
이 코드 예제에서는 AWS SDK로 IAM 사용자를 만들고 사용자에게 정책을 부착하는 방법을 보여드려요.
출처: 문서
본문
다음 작업을 다루어요.
- IAM 사용자 두 명 생성
- 한 사용자에게 Amazon S3 버킷에서 객체를 가져오고 넣을 수 있는 정책 부착
- 두 번째 사용자에게 버킷에서 객체를 가져올 수 있는 정책 부착
- 사용자 자격 증명에 따라 버킷에 대한 서로 다른 권한 적용
Warning 보안 위험을 피하기 위해, 목적 지향 소프트웨어를 개발하거나 실제 데이터를 다룰 때는 인증에 IAM 사용자를 사용하지 마세요. 대신 AWS IAM Identity Center 같은 자격 증명 공급자와의 페더레이션을 사용해요.
Python (SDK for Python / Boto3)
Note GitHub에 더 많은 내용이 있어요. AWS Code Examples Repository에서 전체 예제와 설정·실행 방법을 확인할 수 있어요.
IAM 사용자 동작을 감싸는 래퍼 함수들을 만들어요.
import logging
import time
import boto3
from botocore.exceptions import ClientError
import access_key_wrapper
import policy_wrapper
logger = logging.getLogger(__name__)
iam = boto3.resource("iam")
def create_user(user_name):
"""
Creates a user. By default, a user has no permissions or access keys.
:param user_name: The name of the user.
:return: The newly created user.
"""
try:
user = iam.create_user(UserName=user_name)
logger.info("Created user %s.", user.name)
except ClientError:
logger.exception("Couldn't create user %s.", user_name)
raise
else:
return user
def update_user(user_name, new_user_name):
"""
Updates a user's name.
:param user_name: The current name of the user to update.
:param new_user_name: The new name to assign to the user.
:return: The updated user.
"""
try:
user = iam.User(user_name)
user.update(NewUserName=new_user_name)
logger.info("Renamed %s to %s.", user_name, new_user_name)
except ClientError:
logger.exception("Couldn't update name for user %s.", user_name)
raise
return user
def list_users():
"""
Lists the users in the current account.
:return: The list of users.
"""
try:
users = list(iam.users.all())
logger.info("Got %s users.", len(users))
except ClientError:
logger.exception("Couldn't get users.")
raise
else:
return users
def delete_user(user_name):
"""
Deletes a user. Before a user can be deleted, all associated resources,
such as access keys and policies, must be deleted or detached.
:param user_name: The name of the user.
"""
try:
iam.User(user_name).delete()
logger.info("Deleted user %s.", user_name)
except ClientError:
logger.exception("Couldn't delete user %s.", user_name)
raise
def attach_policy(user_name, policy_arn):
"""
Attaches a policy to a user.
:param user_name: The name of the user.
:param policy_arn: The Amazon Resource Name (ARN) of the policy.
"""
try:
iam.User(user_name).attach_policy(PolicyArn=policy_arn)
logger.info("Attached policy %s to user %s.", policy_arn, user_name)
except ClientError:
logger.exception("Couldn't attach policy %s to user %s.", policy_arn, user_name)
raise
def detach_policy(user_name, policy_arn):
"""
Detaches a policy from a user.
:param user_name: The name of the user.
:param policy_arn: The Amazon Resource Name (ARN) of the policy.
"""
try:
iam.User(user_name).detach_policy(PolicyArn=policy_arn)
logger.info("Detached policy %s from user %s.", policy_arn, user_name)
except ClientError:
logger.exception(
"Couldn't detach policy %s from user %s.", policy_arn, user_name
)
raise
IAM 정책 동작을 감싸는 래퍼 함수들을 만들어요.
import json
import logging
import operator
import pprint
import time
import boto3
from botocore.exceptions import ClientError
logger = logging.getLogger(__name__)
iam = boto3.resource("iam")
def create_policy(name, description, actions, resource_arn):
"""
Creates a policy that contains a single statement.
:param name: The name of the policy to create.
:param description: The description of the policy.
:param actions: The actions allowed by the policy. These typically take the
form of service:action, such as s3:PutObject.
:param resource_arn: The Amazon Resource Name (ARN) of the resource this policy
applies to. This ARN can contain wildcards, such as
'arn:aws:s3:::amzn-s3-demo-bucket/*' to allow actions on all objects
in the bucket named 'amzn-s3-demo-bucket'.
:return: The newly created policy.
"""
policy_doc = {
"Version":"2012-10-17",
"Statement": [{"Effect": "Allow", "Action": actions, "Resource": resource_arn}],
}
try:
policy = iam.create_policy(
PolicyName=name,
Description=description,
PolicyDocument=json.dumps(policy_doc),
)
logger.info("Created policy %s.", policy.arn)
except ClientError:
logger.exception("Couldn't create policy %s.", name)
raise
else:
return policy
def delete_policy(policy_arn):
"""
Deletes a policy.
:param policy_arn: The ARN of the policy to delete.
"""
try:
iam.Policy(policy_arn).delete()
logger.info("Deleted policy %s.", policy_arn)
except ClientError:
logger.exception("Couldn't delete policy %s.", policy_arn)
raise
IAM 액세스 키 동작을 감싸는 래퍼 함수들을 만들어요.
import logging
import boto3
from botocore.exceptions import ClientError
logger = logging.getLogger(__name__)
iam = boto3.resource("iam")
def create_key(user_name):
"""
Creates an access key for the specified user. Each user can have a
maximum of two keys.
:param user_name: The name of the user.
:return: The created access key.
"""
try:
key_pair = iam.User(user_name).create_access_key_pair()
logger.info(
"Created access key pair for %s. Key ID is %s.",
key_pair.user_name,
key_pair.id,
)
except ClientError:
logger.exception("Couldn't create access key pair for %s.", user_name)
raise
else:
return key_pair
def delete_key(user_name, key_id):
"""
Deletes a user's access key.
:param user_name: The user that owns the key.
:param key_id: The ID of the key to delete.
"""
try:
key = iam.AccessKey(user_name, key_id)
key.delete()
logger.info("Deleted access key %s for %s.", key.id, key.user_name)
except ClientError:
logger.exception("Couldn't delete key %s for %s", key_id, user_name)
raise
래퍼 함수들을 사용해 서로 다른 정책을 가진 사용자들을 만들고, 그 자격 증명으로 Amazon S3 버킷에 접근해요.
def usage_demo():
"""
Shows how to manage users, keys, and policies.
This demonstration creates two users: one user who can put and get objects in an
Amazon S3 bucket, and another user who can only get objects from the bucket.
The demo then shows how the users can perform only the actions they are permitted
to perform.
"""
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
print("-" * 88)
print("Welcome to the AWS Identity and Account Management user demo.")
print("-" * 88)
print(
"Users can have policies and roles attached to grant them specific "
"permissions."
)
s3 = boto3.resource("s3")
bucket = s3.create_bucket(
Bucket=f"demo-iam-bucket-{time.time_ns()}",
CreateBucketConfiguration={
"LocationConstraint": s3.meta.client.meta.region_name
},
)
print(f"Created an Amazon S3 bucket named {bucket.name}.")
user_read_writer = create_user("demo-iam-read-writer")
user_reader = create_user("demo-iam-reader")
print(f"Created two IAM users: {user_read_writer.name} and {user_reader.name}")
update_user(user_read_writer.name, "demo-iam-creator")
update_user(user_reader.name, "demo-iam-getter")
users = list_users()
user_read_writer = next(
user for user in users if user.user_id == user_read_writer.user_id
)
user_reader = next(user for user in users if user.user_id == user_reader.user_id)
print(
f"Changed the names of the users to {user_read_writer.name} "
f"and {user_reader.name}."
)
read_write_policy = policy_wrapper.create_policy(
"demo-iam-read-write-policy",
"Grants rights to create and get an object in the demo bucket.",
["s3:PutObject", "s3:GetObject"],
f"arn:aws:s3:::{bucket.name}/*",
)
print(
f"Created policy {read_write_policy.policy_name} with ARN: {read_write_policy.arn}"
)
print(read_write_policy.description)
read_policy = policy_wrapper.create_policy(
"demo-iam-read-policy",
"Grants rights to get an object from the demo bucket.",
"s3:GetObject",
f"arn:aws:s3:::{bucket.name}/*",
)
print(f"Created policy {read_policy.policy_name} with ARN: {read_policy.arn}")
print(read_policy.description)
attach_policy(user_read_writer.name, read_write_policy.arn)
print(f"Attached {read_write_policy.policy_name} to {user_read_writer.name}.")
attach_policy(user_reader.name, read_policy.arn)
print(f"Attached {read_policy.policy_name} to {user_reader.name}.")
user_read_writer_key = access_key_wrapper.create_key(user_read_writer.name)
print(f"Created access key pair for {user_read_writer.name}.")
user_reader_key = access_key_wrapper.create_key(user_reader.name)
print(f"Created access key pair for {user_reader.name}.")
s3_read_writer_resource = boto3.resource(
"s3",
aws_access_key_id=user_read_writer_key.id,
aws_secret_access_key=user_read_writer_key.secret,
)
demo_object_key = f"object-{time.time_ns()}"
demo_object = None
while demo_object is None:
try:
demo_object = s3_read_writer_resource.Bucket(bucket.name).put_object(
Key=demo_object_key, Body=b"AWS IAM demo object content!"
)
except ClientError as error:
if error.response["Error"]["Code"] == "InvalidAccessKeyId":
print("Access key not yet available. Waiting...")
time.sleep(1)
else:
raise
print(
f"Put {demo_object_key} into {bucket.name} using "
f"{user_read_writer.name}'s credentials."
)
read_writer_object = s3_read_writer_resource.Bucket(bucket.name).Object(
demo_object_key
)
read_writer_content = read_writer_object.get()["Body"].read()
print(f"Got object {read_writer_object.key} using read-writer user's credentials.")
print(f"Object content: {read_writer_content}")
s3_reader_resource = boto3.resource(
"s3",
aws_access_key_id=user_reader_key.id,
aws_secret_access_key=user_reader_key.secret,
)
demo_content = None
while demo_content is None:
try:
demo_object = s3_reader_resource.Bucket(bucket.name).Object(demo_object_key)
demo_content = demo_object.get()["Body"].read()
print(f"Got object {demo_object.key} using reader user's credentials.")
print(f"Object content: {demo_content}")
except ClientError as error:
if error.response["Error"]["Code"] == "InvalidAccessKeyId":
print("Access key not yet available. Waiting...")
time.sleep(1)
else:
raise
try:
demo_object.delete()
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDenied":
print("-" * 88)
print(
"Tried to delete the object using the reader user's credentials. "
"Got expected AccessDenied error because the reader is not "
"allowed to delete objects."
)
print("-" * 88)
access_key_wrapper.delete_key(user_reader.name, user_reader_key.id)
detach_policy(user_reader.name, read_policy.arn)
policy_wrapper.delete_policy(read_policy.arn)
delete_user(user_reader.name)
print(f"Deleted keys, detached and deleted policy, and deleted {user_reader.name}.")
access_key_wrapper.delete_key(user_read_writer.name, user_read_writer_key.id)
detach_policy(user_read_writer.name, read_write_policy.arn)
policy_wrapper.delete_policy(read_write_policy.arn)
delete_user(user_read_writer.name)
print(
f"Deleted keys, detached and deleted policy, and deleted {user_read_writer.name}."
)
bucket.objects.delete()
bucket.delete()
print(f"Emptied and deleted {bucket.name}.")
print("Thanks for watching!")
API 상세는 AWS SDK for Python (Boto3) API Reference의 다음 항목을 참고해요.
AttachUserPolicyCreateAccessKeyCreatePolicyCreateUserDeleteAccessKeyDeletePolicyDeleteUserDetachUserPolicyListUsersUpdateUser