#!/usr/bin/env python3 # /// script # requires-python = ">=3.12" # dependencies = [ # "boto3", # ] # /// import argparse import json import random import string import subprocess from enum import Enum from pathlib import Path import boto3 REGION_NAME = "us-east-1" class Actions(str, Enum): Get = "Get" Put = "Put" List = "List" def get_user_dir(bucket_name, user, with_bucket=True): if with_bucket: return f"{bucket_name}/user-id-{user}" return f"user-id-{user}" def create_power_user(): power_user_key = "power_user_key" power_user_secret = "power_user_secret" command = f"s3.configure -apply -user poweruser -access_key {power_user_key} -secret_key {power_user_secret} -actions Admin" print("Creating Power User...") subprocess.run( ["docker", "exec", "-i", "seaweedfs-master-1", "weed", "shell"], input=command, text=True, stdout=subprocess.PIPE, ) print( f"Power User created with key: {power_user_key} and secret: {power_user_secret}" ) return power_user_key, power_user_secret def create_bucket(s3_client, bucket_name): print(f"Creating Bucket {bucket_name}...") s3_client.create_bucket(Bucket=bucket_name) print(f"Bucket {bucket_name} created.") def upload_file(s3_client, bucket_name, user, file_path, custom_remote_path=None): user_dir = get_user_dir(bucket_name, user, with_bucket=False) if custom_remote_path: remote_path = custom_remote_path else: remote_path = f"{user_dir}/{str(Path(file_path).name)}" print(f"Uploading {file_path} for {user}... on {user_dir}") s3_client.upload_file(file_path, bucket_name, remote_path) print(f"File {file_path} uploaded for {user}.") def create_user(iam_client, user): print(f"Creating user {user}...") response = iam_client.create_access_key(UserName=user) print( f"User {user} created with access key: {response['AccessKey']['AccessKeyId']}" ) return response def list_files(s3_client, bucket_name, path=None): if path is None: path = "" print(f"Listing files of s3://{bucket_name}/{path}...") try: response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=path) if "Contents" in response: for obj in response["Contents"]: print(f"\t - {obj['Key']}") else: print("No files found.") except Exception as e: print(f"Error listing files: {e}") def create_policy_for_user( iam_client, user, bucket_name, actions=[Actions.Get, Actions.List] ): print(f"Creating policy for {user} on {bucket_name}...") policy_document = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [f"s3:{action.value}*" for action in actions], "Resource": [ f"arn:aws:s3:::{get_user_dir(bucket_name, user)}/*", ], } ], } policy_name = f"{user}-{bucket_name}-full-access" policy_json = json.dumps(policy_document) filepath = f"/tmp/{policy_name}.json" with open(filepath, "w") as f: f.write(json.dumps(policy_document, indent=2)) iam_client.put_user_policy( PolicyName=policy_name, PolicyDocument=policy_json, UserName=user ) print(f"Policy for {user} on {bucket_name} created.") def main(): parser = argparse.ArgumentParser(description="SeaweedFS S3 Test Script") parser.add_argument( "--s3-url", default="http://127.0.0.1:8333", help="S3 endpoint URL" ) parser.add_argument( "--iam-url", default="http://127.0.0.1:8111", help="IAM endpoint URL" ) args = parser.parse_args() bucket_name = ( f"test-bucket-{''.join(random.choices(string.digits + 'abcdef', k=8))}" ) sentinel_file = "/tmp/SENTINEL" with open(sentinel_file, "w") as f: f.write("Hello World") print(f"SENTINEL file created at {sentinel_file}") power_user_key, power_user_secret = create_power_user() admin_s3_client = get_s3_client(args, power_user_key, power_user_secret) iam_client = get_iam_client(args, power_user_key, power_user_secret) create_bucket(admin_s3_client, bucket_name) upload_file(admin_s3_client, bucket_name, "Alice", sentinel_file) upload_file(admin_s3_client, bucket_name, "Bob", sentinel_file) list_files(admin_s3_client, bucket_name) alice_user_info = create_user(iam_client, "Alice") bob_user_info = create_user(iam_client, "Bob") alice_key = alice_user_info["AccessKey"]["AccessKeyId"] alice_secret = alice_user_info["AccessKey"]["SecretAccessKey"] bob_key = bob_user_info["AccessKey"]["AccessKeyId"] bob_secret = bob_user_info["AccessKey"]["SecretAccessKey"] # Make sure Admin can read any files list_files(admin_s3_client, bucket_name) list_files( admin_s3_client, bucket_name, get_user_dir(bucket_name, "Alice", with_bucket=False), ) list_files( admin_s3_client, bucket_name, get_user_dir(bucket_name, "Bob", with_bucket=False), ) # Create read policy for Alice and Bob create_policy_for_user(iam_client, "Alice", bucket_name) create_policy_for_user(iam_client, "Bob", bucket_name) alice_s3_client = get_s3_client(args, alice_key, alice_secret) # Make sure Alice can read her files list_files( alice_s3_client, bucket_name, get_user_dir(bucket_name, "Alice", with_bucket=False) + "/", ) # Make sure Bob can read his files bob_s3_client = get_s3_client(args, bob_key, bob_secret) list_files( bob_s3_client, bucket_name, get_user_dir(bucket_name, "Bob", with_bucket=False) + "/", ) # Update policy to include write create_policy_for_user(iam_client, "Alice", bucket_name, actions=[Actions.Put, Actions.Get, Actions.List]) # fmt: off create_policy_for_user(iam_client, "Bob", bucket_name, actions=[Actions.Put, Actions.Get, Actions.List]) # fmt: off print("############################# Make sure Alice can write her files") upload_file( alice_s3_client, bucket_name, "Alice", sentinel_file, custom_remote_path=f"{get_user_dir(bucket_name, 'Alice', with_bucket=False)}/SENTINEL_by_Alice", ) print("############################# Make sure Bob can write his files") upload_file( bob_s3_client, bucket_name, "Bob", sentinel_file, custom_remote_path=f"{get_user_dir(bucket_name, 'Bob', with_bucket=False)}/SENTINEL_by_Bob", ) print("############################# Make sure Alice can read her new files") list_files( alice_s3_client, bucket_name, get_user_dir(bucket_name, "Alice", with_bucket=False) + "/", ) print("############################# Make sure Bob can read his new files") list_files( bob_s3_client, bucket_name, get_user_dir(bucket_name, "Bob", with_bucket=False) + "/", ) print("############################# Make sure Bob cannot read Alice's files") list_files( bob_s3_client, bucket_name, get_user_dir(bucket_name, "Alice", with_bucket=False) + "/", ) print("############################# Make sure Alice cannot read Bob's files") list_files( alice_s3_client, bucket_name, get_user_dir(bucket_name, "Bob", with_bucket=False) + "/", ) def get_iam_client(args, access_key, secret_key): iam_client = boto3.client( "iam", endpoint_url=args.iam_url, region_name=REGION_NAME, aws_access_key_id=access_key, aws_secret_access_key=secret_key, ) return iam_client def get_s3_client(args, access_key, secret_key): s3_client = boto3.client( "s3", endpoint_url=args.s3_url, region_name=REGION_NAME, aws_access_key_id=access_key, aws_secret_access_key=secret_key, ) return s3_client if __name__ == "__main__": main()