Python: Implementing a taxonomy based access control - first part

Permission management is one of the things my side project lacks most to become fully operational. As for now it was only about being connected or not being connected, but it is changing.

Screenshot by the author, permissions are created by assigning allowed actions to a resource path.

Please take note that the provided source code is not a final version and might be changed in the future to fix some potential issues.

The inspiration: AWS permission management

Resource and action based permission management

When you set permissions on AWS, it is mostly about allowing actions to a resource identifier (Amazon Resource Names ARNs). This resource can contains wildcard characters (*) to indicate any value for a path part.

{
   "Version": "2012-10-17",
    "Statement": [
    {
        "Sid": "VisualEditor0",
        "Effect": "Allow",
        "Action": "s3:ListAllMyBuckets",
        "Resource": "arn:aws:s3:::*"
    }
  ]
}

The action to list all buckets is given for all s3 resources.

The first part : how to extract the resources allowed for a given rule?

The idea here is not to check if the permissions permit to access a certain resource, but what resources are accessible for a given rule.

I will provide you with some examples of what the code do.

Please notice that you shouldn’t consider the resources list in the examples as the full list of existing resources, but as resources that are allowed by the provided permissions.

First case, extract only one variable part

Without a wildcard in the tested resources:

resources = [
    "data/mongodb/users",
    "data/mongodb/user_group",
    "data/mongodb/groups",
    "data/elastic/arxiv2"
]

to_check = "data/<str:provider>/<str:collection>"

matcher = build_policy_matcher(to_check)

print(matcher(resources))

# > {'mongodb': {'users', 'user_group', 'groups'}, 'elastic': {'arxiv2'}}

For the “collection”, we allow ‘groups’, ‘user_group’ and ‘users’ values. This set can be used to construct filters for a database query (instead of filtering results after fetching them).

With a wildcard in one of the values:

resources = [
    "data/mongodb/users",
    "data/mongodb/*",
    "data/mongodb/groups",
]

to_check = "data/mongodb/<str:collection>"

matcher = build_policy_matcher(to_check)

print(matcher(resources))

# > '*'

In this case we will return the ‘*’ wildcard to indicate that we do not need to apply a filter

With two variable parts

For the ‘mongodb’ provider value we allow values ‘users’, ‘user_group’, ‘groups’. As for the elastic value only ‘arxiv2’ is allowed.

A bit of source code:

from typing import Dict, Set, Any, List

def _build_part_matcher(part: str):
    # helper function to provide a function to handle a path part
    if part.startswith("<") and part.endswith(">"):
        # TODO: handle type casting (int, float, str)
        def part_matcher(other_part: str) -> str | bool:
            return other_part

        return True, part_matcher

    else:
        def part_matcher(other_part: str) -> str | bool:
            return other_part == "*" or other_part == part

        return False, part_matcher

def build_policy_matcher(rule: str):
    # method to build a policy matcher function

    parts = rule.split("/")
    part_matchers = []

    part_extractor_count = 0
    # we construct the part_matchers list
    for part in parts:
        acc, matcher = _build_part_matcher(part)

        part_extractor_count += 1 if acc else 0
        part_matchers.append(matcher)

    def policy_matcher(resource: str) -> bool | List[Any]:
        # helper method that will return False if the resource does not match the rule,
        # else it will provide a list with the matched parts

        extracted_parts: List[Any] = []

        if resource == "*":
            resource = "/".join(["*"] * len(part_matchers))

        for resource_part, matcher_part in zip(resource.split("/"), part_matchers):
            if matcher_part and not resource_part:
                return False
            if resource_part and not matcher_part:
                return extracted_parts

            part_match = matcher_part(resource_part)

            if part_match is True:
                continue
            if part_match is False:
                return False

            extracted_parts.append(part_match)
        return extracted_parts

    def matching_parts(resources: List[str]) -> str | Set[Any] | Dict[str, Any]:
        # helper method to get a usable representation of allowed values
        # from a list of allowed resources (with or without wildcards)

        if part_extractor_count == 1:  # if we have only one placeholder, we return a set (or the '*' value)
            matched_parts_set: Set[str] = set()
            for rsc in resources:
                matched_parts = policy_matcher(rsc)

                if isinstance(matched_parts, list):
                    matched_parts_set.add(matched_parts[0])

            return "*" if "*" in matched_parts_set else matched_parts_set
        
        # else we will construct the final value as a dict > dict >... set
        matched_parts_accumulator: List[List[str]] = []
        for rsc in resources:
            matched_parts = policy_matcher(rsc)

            if isinstance(matched_parts, list):
                matched_parts_accumulator.append(matched_parts)

        final_value: Dict[Any, Any] = {}

        for accumulated in matched_parts_accumulator:
            work_value: Dict[Any, Any] | Set[Any] = final_value
            for idx in range(part_extractor_count):
                is_last = idx + 1 == part_extractor_count
                is_before_last = idx + 2 == part_extractor_count
                part = accumulated[idx]

                if is_last:
                    if part == "*":
                        work_value.clear()
                        work_value.add("*")
                    if "*" not in work_value:
                        work_value.add(part)

                elif is_before_last:
                    work_value = work_value.setdefault(part, set())
                else:
                    work_value = work_value.setdefault(part, {})

        return final_value

    return matching_parts

That’s all for now folks!