py-ngac 0.1.0

Python port of the NIST Policy Machine core for Next Generation Access Control (NGAC).

py-ngac

The core components of the NIST Policy Machine, ported to Python as a reference implementation of the Next Generation Access Control (NGAC) standard.

Installation

Install from PyPI

pip install py-ngac

Install for local development

git clone <your-repo-url>

cd py-ngac

pip install -e .[dev]

or with Poetry:

poetry install

Package Description

• common - Objects common to other packages.
• pap - Policy Administration Point. Provides the Policy Machine implementation of the NGAC PAP interfaces for modifying and querying policy.
• pdp - Policy Decision Point. Implementation of an administrative PDP that controls access to admin operations on the PAP.
• epp - Event Processing Point. The EPP attaches to a PDP to listen to administrative events while exposing an interface for a PEP to send events.
• impl - Policy Machine supported implementations of the PAP interfaces. Included is an in-memory implementation and extension points for additional backends.

The pap package allows you to create NGAC policies without access checks on administrative operations. The pdp is designed to wrap the features of the pap and perform access checks on administrative operations. The epp package provides a means of subscribing to PDP events and processing obligations defined in the PAP.

Getting Started

In general, to start using this library:

from py_ngac.core.epp.epp import EPP
from py_ngac.core.impl.memory.pap.memory_pap import MemoryPAP
from py_ngac.core.pdp.pdp import PDP

pap = MemoryPAP()
pdp = PDP(pap)
epp = EPP(pdp, pap)
epp.subscribe_to(pdp)

Below are two code snippets to show the basics of creating an NGAC policy with the Python port. The first example uses the Python API to create and test a policy. The second defines the policy in PML and tests using Python.

Python

from py_ngac.core.epp.epp import EPP
from py_ngac.core.impl.memory.pap.memory_pap import MemoryPAP
from py_ngac.core.pap.operation.accessright.access_right_set import AccessRightSet
from py_ngac.core.pap.operation.accessright.admin_access_right import AdminAccessRight
from py_ngac.core.pap.operation.admin_operation import AdminOperation
from py_ngac.core.pap.operation.arg.args import Args
from py_ngac.core.pap.operation.arg.type.basic_types import STRING_TYPE, VOID_TYPE
from py_ngac.core.pap.operation.param.formal_parameter import FormalParameter
from py_ngac.core.pap.operation.param.node_name_formal_parameter import NodeNameFormalParameter
from py_ngac.core.pap.operation.reqcap.required_capability import RequiredCapability
from py_ngac.core.pap.operation.reqcap.required_privilege_on_node import RequiredPrivilegeOnNode
from py_ngac.core.pap.operation.reqcap.required_privilege_on_parameter import RequiredPrivilegeOnParameter
from py_ngac.core.pap.operation.resource_operation import ResourceOperation
from py_ngac.core.pap.query.model.context.node_target_context import NodeTargetContext
from py_ngac.core.pap.query.model.context.node_user_context import NodeUserContext
from py_ngac.core.pdp.pdp import PDP
from py_ngac.core.pdp.unauthorized_exception import UnauthorizedException

pap = MemoryPAP()

pap.modify().operations().set_resource_access_rights(AccessRightSet("read", "write"))

pc1_id = pap.modify().graph().create_policy_class("pc1")
users_id = pap.modify().graph().create_user_attribute("users", [pc1_id])
admin_id = pap.modify().graph().create_user_attribute("admin", [pc1_id])
admin_user_id = pap.modify().graph().create_user("admin_user", [admin_id, users_id])
pap.modify().graph().associate(
    admin_id,
    users_id,
    AccessRightSet(AdminAccessRight.ADMIN_GRAPH_ASSIGNMENT_DESCENDANT_CREATE),
)

user_homes_id = pap.modify().graph().create_object_attribute("user homes", [pc1_id])
user_inboxes_id = pap.modify().graph().create_object_attribute("user inboxes", [pc1_id])
pap.modify().graph().associate(admin_id, user_homes_id, AccessRightSet.wildcard())
pap.modify().graph().associate(admin_id, user_inboxes_id, AccessRightSet.wildcard())

pap.modify().prohibitions().create_node_prohibition(
    "deny admin on user inboxes",
    admin_id,
    AccessRightSet("read"),
    {user_inboxes_id},
    set(),
    False,
)

name_param = NodeNameFormalParameter("name")

class ReadFile(ResourceOperation[None]):
    def execute_with_query(self, query, user_ctx, args: Args) -> None:
        _ = query
        _ = user_ctx
        file_name = args.get(name_param)
        print(f'read_file("{file_name}")')
        return None

pap.modify().operations().create_operation(
    ReadFile(
        "read_file",
        VOID_TYPE,
        [name_param],
        [RequiredCapability(RequiredPrivilegeOnParameter(name_param, AccessRightSet("read")))],
    )
)

username_param = FormalParameter("username", STRING_TYPE)

class CreateNewUser(AdminOperation[None]):
    def execute(self, runtime_pap, user_ctx, args: Args) -> None:
        _ = user_ctx
        username = args.get(username_param)
        runtime_pap.modify().graph().create_user(username, [users_id])
        runtime_pap.modify().graph().create_object_attribute(username + " home", [user_homes_id])
        runtime_pap.modify().graph().create_object_attribute(username + " inbox", [user_inboxes_id])
        return None

pap.modify().operations().create_operation(
    CreateNewUser(
        "create_new_user",
        VOID_TYPE,
        [username_param],
        [
            RequiredCapability(
                RequiredPrivilegeOnNode(
                    "users",
                    AdminAccessRight.ADMIN_GRAPH_ASSIGNMENT_DESCENDANT_CREATE,
                )
            )
        ],
    )
)

pap.execute_pml(
    NodeUserContext.of(admin_user_id),
    """
    create obligation "o1"
    when any user
    performs "create_new_user"
    do(ctx) {
        objName := "welcome " + ctx.args.username
        inboxName := ctx.args.username + " inbox"
        create o objName in [inboxName]
    }
    """,
)

pdp = PDP(pap)
epp = EPP(pdp, pap)
epp.subscribe_to(pdp)

pdp.adjudicate_operation(NodeUserContext.of(admin_user_id), "create_new_user", {"username": "testUser"})

assert pap.query().graph().node_exists("testUser home")
assert pap.query().graph().node_exists("testUser inbox")
assert pap.query().graph().node_exists("welcome testUser")

try:
    pdp.adjudicate_operation(NodeUserContext.of("testUser"), "create_new_user", {"username": "testUser2"})
except UnauthorizedException:
    pass

PML

from py_ngac.core.epp.epp import EPP
from py_ngac.core.impl.memory.pap.memory_pap import MemoryPAP
from py_ngac.core.pap.query.model.context.node_user_context import NodeUserContext
from py_ngac.core.pdp.bootstrap.pml_bootstrapper import PMLBootstrapper
from py_ngac.core.pdp.pdp import PDP
from py_ngac.core.pdp.unauthorized_exception import UnauthorizedException

PML = """
set resource access rights ["read", "write"]

create pc "pc1"
create ua "users" in ["pc1"]
create ua "admin" in ["pc1"]
assign "admin_user" to ["admin", "users"]
associate "admin" to "users" with ["admin:graph:assignment:descendant:create"]

create oa "user homes" in ["pc1"]
create oa "user inboxes" in ["pc1"]
associate "admin" to "user homes" with ["*"]
associate "admin" to "user inboxes" with ["*"]

create conj node prohibition "deny admin on user inboxes"
deny "admin"
arset ["read"]
include ["user inboxes"]

@ReqCap({
    require ["read"] on [name]
})
resourceop read_file(@Node string name) { }

@ReqCap({
    require ["admin:graph:assignment:descendant:create"] on ["users"]
})
adminop create_new_user(string username) {
    create u username in ["users"]
    create oa username + " home" in ["user homes"]
    create oa username + " inbox" in ["user inboxes"]
}

create obligation "o1"
when any user
performs "create_new_user"
do(ctx) {
    objName := "welcome " + ctx.args.username
    inboxName := ctx.args.username + " inbox"
    create o objName in [inboxName]
}
"""

pap = MemoryPAP()
pap.bootstrap(PMLBootstrapper("admin_user", PML))

pdp = PDP(pap)
epp = EPP(pdp, pap)
epp.subscribe_to(pdp)

admin_user_id = pap.query().graph().get_node_id("admin_user")
pdp.execute_pml(NodeUserContext.of(admin_user_id), 'create_new_user(username="testUser")')

assert pap.query().graph().node_exists("testUser home")
assert pap.query().graph().node_exists("testUser inbox")
assert pap.query().graph().node_exists("welcome testUser")

try:
    pdp.execute_pml(NodeUserContext.of("testUser"), 'create_new_user(username="testUser2")')
except UnauthorizedException:
    pass

Operations

Operations are a fundamental part of NGAC and this Python port. There are 5 types of supported operations:

• Admin Operations: Modify the policy.
• Resource Operations: Represents access on a resource (object).
• Query Operations: Query the policy information.
• Routines: A set of operations, with access checks on each statement rather than the routine itself or its args.
• Functions: Reusable utility operations that do not access the policy.

Only Admin and Resource operations emit events to the EPP. All operations define a set of formal parameters. Admin, Resource, and Query operations can define required capabilities which are a set of access rights a user needs on the actual argument in the call to the operation in order to successfully execute the operation.

Define an operation

Define an admin and resource operation using PML.

set resource access rights ["read"]

resourceop read_file(@Node("read") string filename) map[string]any {
    return getNode(filename)
}

adminop create_new_user(string username) {
    require ["admin:graph:assignment:descendant:create"] on ["users"]

    create u username in ["users"]
    create oa username + " home" in ["user homes"]
    create oa username + " inbox" in ["user inboxes"]
}

Execute operation

pap.execute_pml(user_ctx, pml)

# or

pdp.execute_pml(user_ctx, pml)

Obligation example

Now that the operations are persisted in the PIP, they can be executed by the PDP and the resulting events can be processed by the EPP against any obligations.

create obligation "o1"
when any user
performs read_file on (filename) {
    return filename == "file1.txt"
}
do(ctx) {
    # do something
}

Now when the PDP executes an operation:

pdp.adjudicate_operation(user_ctx, "read_file", {"filename": "file1.txt"})

It will emit an event:

{
  "user": "<username>",
  "process": "<process>",
  "opName": "read_file",
  "args": {
    "filename": "file1.txt"
  }
}

Which matches the obligation's operation pattern and the EPP will execute the obligation response.

Policy Machine Language (PML)

PML is a domain-specific language for defining NGAC access control policies. It provides a declarative syntax that is easier to use and maintain than building everything through the Python API.

JSON Serialization

Policies can be exported and imported using the JSON serializer and deserializer included in the port.

from py_ngac.core.impl.memory.pap.memory_pap import MemoryPAP
from py_ngac.core.pap.serialization.json.json_deserializer import JSONDeserializer
from py_ngac.core.pap.serialization.json.json_serializer import JSONSerializer

json_payload = pap.serialize(JSONSerializer())

new_pap = MemoryPAP()
new_pap.deserialize(json_payload, JSONDeserializer())

Custom Policy Store Implementations

An in-memory implementation of the PAP and PolicyStore interfaces is provided, along with extension points for additional backends.

To implement a custom policy store:

1. Implement the PolicyStore protocol or interface family.

class CustomPolicyStore:
    # implementation details...
    pass

2. Implement required store interfaces.

• GraphStore
• ProhibitionsStore
• ObligationsStore
• OperationsStore

3. Create a custom PAP backed by the store.

See the memory implementation under src/py_ngac/core/impl/memory/pap for a complete example.

Licensing

This repository is a modified derivative work and Python port of the NIST policy-machine-core project.

• See LICENSE.md
• See NOTICE.md