Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use untag_role method in localstack

Best Python code snippet using localstack_python

handler.py

Source:handler.py

1import sys2import traceback3import ujson as json4from asgiref.sync import sync_to_async5from botocore.exceptions import ClientError6from cloudaux.aws.sts import boto3_cached_conn7from consoleme.config import config8from consoleme.lib.aws import sanitize_session_name9from consoleme.lib.plugins import get_plugin_by_name10from consoleme.lib.role_updater.schemas import RoleUpdaterRequest11log = config.get_logger()12stats = get_plugin_by_name(config.get("plugins.metrics", "default_metrics"))()13async def update_role(event):14    log_data = {15        "function": f"{__name__}.{sys._getframe().f_code.co_name}",16        "event": event,17        "message": "Working on event",18    }19    log.debug(log_data)20    if not isinstance(event, list):21        raise Exception("The passed event must be a list.")22    # Let's normalize all of the policies to JSON if they are not already23    for d in event:24        for i in d.get("inline_policies", []):25            if i.get("policy_document") and isinstance(i.get("policy_document"), dict):26                i["policy_document"] = json.dumps(27                    i["policy_document"], escape_forward_slashes=False28                )29        if d.get("assume_role_policy_document", {}):30            if isinstance(31                d.get("assume_role_policy_document", {}).get(32                    "assume_role_policy_document"33                ),34                dict,35            ):36                d["assume_role_policy_document"][37                    "assume_role_policy_document"38                ] = json.dumps(39                    d["assume_role_policy_document"]["assume_role_policy_document"],40                    escape_forward_slashes=False,41                )42    bad_validation = RoleUpdaterRequest().validate(event, many=True)43    if bad_validation:44        log_data["error"] = bad_validation45        log.error(log_data)46        return {"error_msg": "invalid schema passed", "detail_error": bad_validation}47    event = RoleUpdaterRequest().load(event, many=True)48    result = {"success": False}49    for d in event:50        arn = d["arn"]51        aws_session_name = sanitize_session_name("roleupdater-" + d["requester"])52        account_number = await parse_account_id_from_arn(arn)53        role_name = await parse_role_name_from_arn(arn)54        # TODO: Make configurable55        client = boto3_cached_conn(56            "iam",57            account_number=account_number,58            assume_role=config.get("policies.role_name", "ConsoleMe"),59            session_name=aws_session_name,60            retry_max_attempts=2,61            client_kwargs=config.get("boto3.client_kwargs", {}),62        )63        inline_policies = d.get("inline_policies", [])64        managed_policies = d.get("managed_policies", [])65        assume_role_doc = d.get("assume_role_policy_document", {})66        tags = d.get("tags", [])67        if (68            not inline_policies69            and not managed_policies70            and not assume_role_doc71            and not tags72        ):73            result["message"] = f"Invalid request. No response taken on event: {event}"74            return result75        try:76            for policy in inline_policies:77                await update_inline_policy(client, role_name, policy)78            for policy in managed_policies:79                await update_managed_policy(client, role_name, policy)80            if assume_role_doc:81                await update_assume_role_document(client, role_name, assume_role_doc)82            for tag in tags:83                await update_tags(client, role_name, tag)84        except ClientError as ce:85            result["message"] = ce.response["Error"]86            result["Traceback"] = traceback.format_exc()87            return result88        result["success"] = True89        return result90async def parse_account_id_from_arn(arn):91    return arn.split(":")[4]92async def parse_role_name_from_arn(arn):93    return arn.split("/")[-1]94async def update_inline_policy(client, role_name, policy):95    log.debug(96        {"message": "Updating inline policy", "role_name": role_name, "policy": policy}97    )98    if policy.get("action") == "attach":99        response = await sync_to_async(client.put_role_policy)(100            RoleName=role_name,101            PolicyName=policy["policy_name"],102            PolicyDocument=policy["policy_document"],103        )104    elif policy.get("action") == "detach":105        response = await sync_to_async(client.delete_role_policy)(106            RoleName=role_name, PolicyName=policy["policy_name"]107        )108    else:109        raise Exception("Unable to update managed policy")110    return response111async def update_managed_policy(client, role_name, policy):112    log.debug(113        {"message": "Updating managed policy", "role_name": role_name, "policy": policy}114    )115    if policy.get("action") == "attach":116        response = await sync_to_async(client.attach_role_policy)(117            PolicyArn=policy["arn"], RoleName=role_name118        )119    elif policy.get("action") == "detach":120        response = await sync_to_async(client.detach_role_policy)(121            PolicyArn=policy["arn"], RoleName=role_name122        )123    else:124        raise Exception("Unable to update managed policy.")125    return response126async def update_assume_role_document(client, role_name, assume_role_doc):127    log.debug(128        {129            "message": "Updating assume role doc",130            "role_name": role_name,131            "assume_role_doc": assume_role_doc,132        }133    )134    response = None135    if assume_role_doc.get("action", "") in ["create", "update"]:136        response = await sync_to_async(client.update_assume_role_policy)(137            RoleName=role_name,138            PolicyDocument=assume_role_doc["assume_role_policy_document"],139        )140    return response141    # Log or report result?142async def update_tags(client, role_name, tag):143    log.debug({"message": "Updating tag", "role_name": role_name, "tag": tag})144    if tag.get("action") == "add":145        response = await sync_to_async(client.tag_role)(146            RoleName=role_name, Tags=[{"Key": tag["key"], "Value": tag["value"]}]147        )148    elif tag.get("action") == "remove":149        response = await sync_to_async(client.untag_role)(150            RoleName=role_name, TagKeys=[tag["key"]]151        )152    else:153        raise Exception("Unable to update tags.")...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.