Blog

Using undocumented AWS APIs

21 Oct, 2023
Xebia Background Header Wave

TL;DR just give me the code

While evaluating some existing IAM policies in a codebase, I found myself repeating the same steps over and over again: navigate Google and search iam actions servicename and look up information about the actions used.

Pro tip: it is much easier to just bookmark this one: https://docs.aws.amazon.com/service-authorization/latest/reference/reference_policies_actions-resources-contextkeys.html

The information I needed for my policy validation work is quite simple:

  • what are all the services?

  • what are all the actions a service has?

  • what actions match a pattern like “Desc*” for a service?

  • what are mandatory and optional resources for each action?

  • what condition keys can be used?

  • is the action readonly, readwrite or something else?

The first place to go to find out if this information is somehow exposed would be the AWS SDKs. I looked through the boto3 documentation on the iam service and came up empty.

After doing a lot of policies using the manual process, I remembered that AWS has a policy-editing tool in the console that seems to be using the information I was looking up manually. So I started my adventure by trying to automate my struggles.

I decided to invest some time in the policy editor, which was using some kind of API I could use to automate some things. So with the Chrome Inspect pane using the network tab, I saw a lot of http requests to:

https://us-east-1.console.aws.amazon.com/iamv2/api/iamv2

After some experimentation and fiddling with cookies and CSRF tokens, I found out how this undocumented API worked. So I cooked up a little Python to automate that. Since it is only 80 lines of code, I’ll share it here. I will probably make an installable Python package of it soon. The repository with the code and some examples is https://github.com/binxio/aws-iamv2.

import requests
import json
import boto3
from bs4 import BeautifulSoup

# composePolicy, decomposePolicy, checkMultiMFAStatus, createX509, cuid, generateKeyPairs
methods = { "services"                    : lambda p: None,
            "actions"                     : lambda p: { "serviceName": p, "RegionName": "eu-central-1" },
            "resources"                   : lambda p: None,
            "contextKeys"                 : lambda p: None,
            "globalConditionKeys"         : lambda p: None,
            "getServiceLinkedRoleTemplate": lambda p: { "serviceName": p },
            "policySummary"               : lambda p: { "policyDocument": p },
            "validate"                    : lambda p: { "policy": json.dumps(p), "type": "" } }

class ConsoleSession:
    def __init__(self, boto3_session):
        self._credentials = boto3_session.get_credentials()
        self._signed_in = False
        self._csrf_token = None
        self._cache = {method: {} for method in methods}
        self._rsession = requests.Session()

    def __getattribute__(self, name):
        if name in methods:
            def make_lambda(method, converter):
                return lambda param=None: self.get_api_result(method, converter(param))
            return make_lambda(name, methods[name])
        else:
            return object.__getattribute__(self, name)

    def signin(self):
        token = json.loads(self._rsession.get(
            "https://signin.aws.amazon.com/federation", 
            params={
                "Action": "getSigninToken",
                "Session": json.dumps({
                    "sessionId": self._credentials.access_key,
                    "sessionKey": self._credentials.secret_key,
                    "sessionToken": self._credentials.token
                })
            }
        ).text)["SigninToken"]
        self._rsession.get(
            "https://signin.aws.amazon.com/federation",
            params={
                "Action": "login",
                "Issuer": None,
                "Destination": "https://console.aws.amazon.com/",
                "SigninToken": token
            }
        )
        for m in BeautifulSoup(self._rsession.get(
            "https://us-east-1.console.aws.amazon.com/iamv2/home#",
            params={ "region": "eu-central-1", "state": "hashArgs" }
        ).text, "html.parser").find_all("meta"):
            if m.get("name") == "awsc-csrf-token":
                self._csrf_token = m["content"]
        self._signed_in = True

    def get_api_result(self, path, param=None):
        not self._signed_in and self.signin()
        params = json.dumps(param)
        if self._cache[path].get(params, None):
            return self._cache[path][params]
        self._cache[path][params] = json.loads(self._rsession.post(
            "https://us-east-1.console.aws.amazon.com/iamv2/api/iamv2",
            headers={
                "Content-Type": "application/json",
                "X-CSRF-Token": self._csrf_token,
            },
            data=json.dumps({
                "headers": { "Content-Type": "application/json" },
                "path": f"/prod/{path}",
                "method": "POST",
                "region": "us-east-1",
                "params": {},
                **({ "contentString": params } if params else {})
            })
        ).text)
        return self._cache[path][params]

A few things about this code:

  • The ConsoleSession class takes a boto3.Session as input. This session needs no actual rights to AWS.

  • The code might look a bit strange because I wanted to make it as a dynamic class so I only had to add one line to implement an extra API endpoint. This uses the __getattribute__ override and the methods object.

  • I use the requests module and start a requests.Session() that does much of the heavy lifting handling cookies needed for the http requests to succeed.

  • To fetch the awsc-csrf-token from the page I use BeautifulSoup

  • The method in methods were not all discover by me, I did a search on iamv2 on github and found some json files that were already detailing this API.

  • Since the information retrieved for the API calls I needed does not change per request I implemented a simple caching feature. For some methods like policySummary and validate this might not be optimal.

The actual signing in is done using three HTTP requests:

  1. Getting a SigninToken from signin.aws.amazon.com/federation

  2. Use the token to login to https://console.aws.amazon.com/

  3. Retrieve https://us-east-1.console.aws.amazon.com/iamv2/home# to get the awsc-csrf-token from the page.

Example usage

The code below demonstrates example usage:

import boto3
from iamv2 import ConsoleSession
import re

awssvcs = {}
console_session = None

def get_iam_info():
    global console_session
    boto_session = boto3.Session(region_name='us-east-1')
    console_session = ConsoleSession(boto_session)

    services = console_session.services()
    for service in services:
        name = service["serviceName"]
        if name not in awssvcs:
            awssvcs[name] = { "parts": [] }
        awssvcs[name]["parts"].append(service)

def get_statement_actions(statement):
    result = []
    actions = statement.get("Action") or statement.get("NotAction")
    reverse = "NotAction" in statement
    reverse = not reverse if statement["Effect"] == "Deny" else reverse
    actions = [actions] if isinstance(actions, str) else actions
    for action in actions:
        service, act = action.split(':')
        if "Actions" not in awssvcs[service]:
            awssvcs[service]["Actions"] = console_session.actions(awssvcs[service]["parts"][0]["serviceKeyName"])
        actrgx = act.replace('*', '[A-Za-z]+')
        for svc_action in awssvcs[service]["Actions"]:
            if bool(re.match(actrgx, svc_action["actionName"], flags=re.IGNORECASE)) ^ reverse:
                result.append(svc_action)
    return result

def get_policy_actions(policy):
    for statement in policy["Statement"]:
        yield get_statement_actions(statement)

get_policy_actions will simply list all the actions allowed by the statements in the policy. Here it is in action:

    policy = {
        "Version": "2012-10-17", 
        "Statement": [{
            "Sid": "ReadOnlyCloudTrail",
            "Effect": "Deny", 
            "NotAction": "cloudtrail:De*", 
            "Resource": "*"
        }]
    }

    get_iam_info()
    for statement_actions in get_policy_actions(policy):
        statement_actions = sorted(statement_actions, key=lambda x: x["actionName"])
        for action in statement_actions:
            print(f'{action["actionName"]:40} {", ".join(action["actionGroups"])}')

This will give the following list:

DeleteChannel                            ReadWrite
DeleteEventDataStore                     ReadWrite
DeleteResourcePolicy                     ReadWrite
DeleteServiceLinkedChannel               ReadWrite
DeleteTrail                              ReadWrite
DeregisterOrganizationDelegatedAdmin     ReadWrite
DescribeQuery                            ReadOnly, ReadWrite
DescribeTrails                           ReadOnly, ReadWrite

You can see that this policy allows some actions that are not read-only. You could use this code in tests being evaluated in your pipeline to make sure you never accidentally allow non-readonly actions in this policy.

Future

I will make an installable Python package from the API part. Also, I have some ideas for some neat policy tools:

  • check for common mistakes in policies.

  • generate readonly service statements for in a permissions boundary

Jacco Kulman
Jacco is a Cloud Consultant at Binx.io. As an experienced development team lead he coded for the banking- and hospitality- and media-industries. He is a big fan of serverless architectures. In his free time he reads science fiction, contributes to open source projects and enjoys being a life-long-learner.
Questions?

Get in touch with us to learn more about the subject and related solutions

Explore related posts