-
Notifications
You must be signed in to change notification settings - Fork 173
Library Reference
The principalmapper
module exposes several different classes and functions and can be used for your own purposes. This page details the different submodules, classes, and functions to enable you to use Principal Mapper as a library.
The common
submodule has all of the different main classes for working with Graphs and OrganizationTree objects. The classes are exposed through __all__
, meaning you can do:
from principalmapper.common import Graph, Node, Edge
The Graph is an object containing Nodes, Edges, Groups, and Policies along with some metadata (AWS account ID, PMapper version). You can create a Graph a few ways:
from principalmapper.common import Graph, Node, Edge, Policy, Group
# to construct Graph you need all the nodes/edges/policies/groups/metadata ready to go
# nodes: List[Node]
# edges: List[Edge]
# policies: List[Policy]
# groups: List[Group]
# metadata: dict, must have keys 'account_id' and 'pmapper_version' or ValueError is raised
g1 = Graph(nodes, edges, policies, groups, metadata)
# if you have a Graph stored on-disk, there's a classmethod to load it
g2 = Graph.create_graph_from_local_disk('graph_dir')
# And with a Graph object you can drop it to disk with a method
g1.store_graph_as_json('other_graph_dir')
Nodes represent IAM Users and Roles. Here's the signature of the constructor:
def __init__(self, arn: str, id_value: str, attached_policies: Optional[List[Policy]], group_memberships: Optional[List[Group]], trust_policy: Optional[dict], instance_profile: Optional[List[str]], num_access_keys: int, active_password: bool, is_admin: bool, permissions_boundary: Optional[Union[str, Policy]], has_mfa: bool, tags: Optional[dict]):
These are fairly self-explanatory. The constructor itself has a bunch of checks to make sure that all the values are filled and reasonable (i.e. you can't set a trust document for an IAM User). All of those input parameters can be read and updated after construction, such as is_admin
.
Additionally, Node objects contain a cache
dictionary. The cache is used for preventing recomputations, such as finding Edges where the Node object is the source of the Edge (get_outbound_edges(graph: Graph)
method) or giving the "searchable name" of the Node (searchable_name()
method).
Edges represent how one Node can authenticate as another Node, thus gaining the ability to make AWS API calls with that other Node's permissions. They are constructed with:
def __init__(self, source: Node, destination: Node, reason: str, short_reason: str)
The reason
/short_reason
parameters describe why the source node can access the destination node. PMapper's current code just lists a service name for the short_reason
field. There is also a method, describe_edge()
that returns a string that chains the source/reason/destination in a way that usually formulates a complete sentence. This is seen when running query
where a principal has to access another principal to make an API call.
Groups correspond to IAM Groups. IAM Users can belong to one or more IAM Groups, IAM Groups can have multiple IAM Users as members. Groups can have attached policies, which filter down to its member IAM Users. Group objects in PMapper are constructed like so:
def __init__(self, arn: str, attached_policies: Optional[List[Policy]]):
Policies define the permissions that a given principal has, and are checked during the authorization process. These policies are defined using JSON objects. PMapper needs them serialized as dictionaries before passing it as a constructor:
def __init__(self, arn: str, name: str, policy_doc: dict):
The arn
parameter should either be the ARN of the managed policy being represented, or the ARN of the resource in the case of resource policies.
OrganizationTree objects represent an organization from AWS Organizations, as well as its hierarchy of OUs (represented with OrganizationNode objects). These can be constructed as well as loaded and saved to disk as Graphs are:
def __init__(self, org_id: str, management_account_id: str, root_ous: List[OrganizationNode], all_scps: List[Policy], accounts: List[str], edge_list: List[Edge], metadata: dict):
from principalmapper.common import OrganizationTree
org1 = OrganizationTree.create_from_dir('org-dir')
org1.save_organization_to_disk('alternate-org-dir')
For manual construction, see principalmapper.gathering.get_organizations_data
for an example of the traversals needed to build the OU hierarchy.
The graphing
submodule contains several files for generating Graph objects (including contained Node/Edge/Group/Policy objects) by interacting with the AWS API. The important submodules are:
graph_actions
contains functions for generating and accessing Graph objects, and gets used by the graph
subcommand of the PMapper CLI. In particular:
-
create_new_graph(session: botocore.session.Session, service_list: List[str], region_allow_list: Optional[List[str]] = None, region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None)
: this takes a botocore session (seeprincipalmapper.util.botocore_tools
for getting one) and a list of services (seechecker_map
fromprincipalmapper.graphing.edge_identification
) to call the AWS API and compile the data needed to produce a Graph. -
create_graph_from_disk(location: str)
: this wraps around the Graph classmethodcreate_graph_from_local_disk
and creates a Graph from on-disk data.
The various submodules with the suffix _edges
produce Edge objects. They contain a class that is a child of EdgeChecker
which expects a return_edges
method implementation. Some of this requires pulling data from additional services (CloudFormation, Lambda), thus the EdgeChecker
constructor requires a botocore session as well.
These submodules also have functions named generate_edges_locally
, which create Edge objects based on input data rather requiring calls to the AWS API. This enables you to create Edges (and Graphs) based on infrastructure-as-code rather than calling the AWS API for an already-implemented system. One of the future goals of this tool is to be incorporated earlier in the development/operations process and catch risks before they make it to real infrastructure.
To run queries programmatically, you should use the query_interface
submodule. It contains functions both for testing if a single principal can make a given API call as well as if the principal can pivot to other principals to make an API call.
This function does full policy evaluation for a single principal, no pivot checks. The arguments are:
-
principal: Node
- The Node representing the IAM User/Role being tested for authorization -
action_to_check: str
- The action being tested for (such ass3:CreateBucket
orec2:RunInstances
) -
resource_to_check: str
- The resource being tested for. For wildcards (*
) you have to specify that asterisk in a string. -
condition_keys_to_check: dict
- The different condition context keys and their values to apply during authorization. Note that a bunch of these get automatically inferred (current time, principal account, principal arn, principal tags) when calling this function. See the_infer_condition_keys
function for details. -
resource_policy: Optional[dict] = None
- When specified, includes the given resource policy (serialized and passed as a dictionary) as part of the authorization check. Note that if this is notNone
, you have to specify theresource_owner
parameter, this function will not make that inference. -
resource_owner: Optional[str] = None
- When specified, marks which account (by ID) owns the resource to which a resource policy is applied. This must be specified ifresource_policy
is specified. -
service_control_policy_groups: Optional[List[List[Policy]]] = None
- This parameter includes SCPs in the authorization check. Note the format is a list of a list. The overarching list is the groups of SCPs that apply as you traverse from a root OU to the account's OU with the account's SCP at the tail end. The inner lists are the collection of SCPs at each level of the traversal. There is a function calledproduce_scp_list
in thequery_orgs
submodule that produces the correct value to pass as this argument, it needs the Graph and OrganizationTree objects to work. -
session_policy: Optional[dict] = None
- This parameter sets a session policy for the authorization check, serialized as a dictionary.
The function returns a bool
indicating that the request would be authorized or not. It also works for inter-account access checks where resource policies allow it.
This function has a few sibling functions with similar signatures:
-
local_check_authorization
: only checks the caller's IAM policies + permission boundary, but does not support session policies/SCPs and cross-account testing. -
local_check_authorization_handling_mfa
: callslocal_check_authorization_full
to start. If that returns False, it applies condition keys for MFA to see if having MFA allows the request and callslocal_check_authorization_full
again. This returns a(bool, bool)
tuple, the first says whether or not the principal was authorized and the second says if MFA is required.
This function is similar to local_check_authorization_full
in signature, but it takes an additional parameter in the first spot. That additional parameter is the Graph object for the account. It also returns a different response value, a QueryResult object. This response expresses how an IAM User/Role could pivot to other principals to make an API request if they aren't authorized to do it to begin with.
QueryResult objects (query_result
submodule) are constructed with:
def __init__(self, allowed: bool, edge_list: Union[List[Edge], Node], node: Node):
The allowed
field says whether or not the principal was authorized or able to pivot to get to an authorized principal. The edge_list
field is either:
- A list of Edge objects, representing the path a principal would have to traverse to make an authorized request
- A Node object. If it's a Node object, that means the Node was an admin that could not directly make the request, but by nature of being an admin could just assign themselves permission and make the request
The node
field represents the caller the authorization check was working from.
The util
submodule contains extra code used in other submodules, which may also be useful for your code using PMapper.
The arns
submodule contains functions to get the different components of an ARN, which all take a single str
argument and return a str
. The functions are:
get_partition
get_service
get_region
get_account_id
-
get_resource
: This is the trailing part of the ARN. Some services separate with colons for different types of resources, some separate with forward slashes. Either way, this returns all of it. -
validate_arn
: This actually returnsbool
to indicate if the passedstr
looks like an ARN.
This submodule has two functions:
-
get_session(profile_arg: Optional[str])
: This function returns a botocoreSession
object and attempts to invokests:GetCallerIdentity
to validate, which raises an error if it cannot. If you specify a value forprofile_arg
, it creates a session from that profile. Otherwise, it'll go through environment variables/instance metadata/etc. as implemented bybotocore.session.get_session()
. -
get_regions_to_search(session: botocore.session.Session, service_name: str, region_allow_list: Optional[List[str]] = None, region_deny_list: Optional[List[str]])
: Given a botocoreSession
object, the name of a service, and either an allow-list or deny-list (but not both), this function returns a list of regions that this service supports (union allow-list or minus deny-list).
The storage
submodule has one valuable function called get_storage_root
. It provides the default location that PMapper stores Graph/Organization data on-disk. The returned value varies depending on the operating system that Python reports (via sys.platform
) but can also be overridden by setting the PMAPPER_STORAGE
environment variable.