Quick start

Per questo quick-start usiamo un deployment gia' pronto installato come descritto qui. Raggiungibile a https://131.154.97.121:9001/

  • Setup Oidc-agent as describe here
  • cd examples && pip3 install --user -r requirements.txt
  • curl https://rclone.org/install.sh | sudo bash
  • TOKEN=oidc-token demopython3 s3_api.py
    • this will use the id_token provided retrieve minio credentials
    • check if the user bucket exists and create it if not
    • do some operation with S3 python api and exit

Code walk-through

First of all you have to get your IAM token from oidc-agent and put it into TOKEN env variable.

export TOKEN=`oidc-token demo`

then in the python code you can recall it with:

import os

token = os.getenv("TOKEN")

Now you need to use this token to retrieve your Minio S3 temporary credentials. One way to do this is as follows:

import requests
import xmltodict


r = requests.post("https://131.154.97.121:9001",
                  data={
                      'Action':
                      "AssumeRoleWithWebIdentity",
                      'Version': "2011-06-15",
                      'WebIdentityToken': token,
                      'DurationSeconds': 900
                  },
                  verify='MINIO.pem')

print(r.status_code, r.reason)

tree = xmltodict.parse(r.content)

credenstials = dict(tree['AssumeRoleWithWebIdentityResponse']
                    ['AssumeRoleWithWebIdentityResult']['Credentials'])

You have now a set of valid credentials composed by an accessID, a secret, and a session token:

print("accessID:", credenstials['AccessKeyId'])
print("secret:", credenstials['SecretAccessKey'])
print("session token:", credenstials['SessionToken'])

Now before mounting your bucket with RClone, you might want to check if the bucket with your username exists, and in case not, to create one. You can do it easily with the MINIO python APIs:

import jwt
from minio import Minio
from minio.error import (ResponseError, BucketAlreadyOwnedByYou,
                         BucketAlreadyExists)

# Get username from your token
username = jwt.decode(token, verify=False)[
    'preferred_username'].lower().split("@")[0]

# Prepare the local directory, where you will mount your bucket
directory = "/tmp/" + username
if not os.path.exists(directory):
    os.makedirs(directory)

# Initialize Minio client
minioClient = Minio(
    '131.154.97.121:9001',
    access_key=credenstials['AccessKeyId'],
    secret_key=credenstials['SecretAccessKey'],
    session_token=credenstials['SessionToken'],
    secure=True,
    http_client=urllib3.PoolManager(
        timeout=urllib3.Timeout.DEFAULT_TIMEOUT,
        cert_reqs='CERT_REQUIRED',
        ca_certs="MINIO.pem",
    )
)

# Make a bucket with the make_bucket API call. Skip if already exists
try:
    minioClient.make_bucket(username)
except BucketAlreadyOwnedByYou:
    pass
except BucketAlreadyExists:
    pass
except ResponseError as err:
    raise err

Now it's all set, you can start to leverage MINIO client for doing whatever you like.

Listing buckets:

buckets = minioClient.list_buckets()

for bucket in buckets:
    print(bucket.name, bucket.creation_date)

Loading a file on a subdirectory of your bucket:

import uuid
#uniq file name just for lazy testing
filename = "%s.txt" % uuid.uuid1()

#put a object in a sub-directory of the bucket
try:
    with open('requirements.txt', 'rb') as file_data:
        file_stat = os.stat('requirements.txt')
        minioClient.put_object( username, 'my_object/test_objec_%s' % filename, file_data, file_stat.st_size)
except ResponseError as err:
    print(err)

Streaming data into MINIO:

# streaming data to a directory within the bucket
data = "I want to stream some test to minio"
data_bytes = data.encode('utf-8')
data_stream = io.BytesIO(data_bytes)

try:
    minioClient.put_object(username, "my_stream/test_stream_%s" % filename, data_stream , len(data_bytes))
except Exception as ex:
    raise ex