mirror of
https://github.com/meta-llama/llama-stack.git
synced 2025-06-28 10:54:19 +00:00
7c1998db25
This allows a set of rules to be defined for determining access to
resources. The rules are (loosely) based on the cedar policy format.
A rule defines a list of action either to permit or to forbid. It may
specify a principal or a resource that must match for the rule to take
effect. It may also specify a condition, either a 'when' or an 'unless',
with additional constraints as to where the rule applies.
A list of rules is held for each type to be protected and tried in order
to find a match. If a match is found, the request is permitted or
forbidden depening on the type of rule. If no match is found, the
request is denied. If no rules are specified for a given type, a rule
that allows any action as long as the resource attributes match the user
attributes is added (i.e. the previous behaviour is the default.
Some examples in yaml:
```
model:
- permit:
principal: user-1
actions: [create, read, delete]
comment: user-1 has full access to all models
- permit:
principal: user-2
actions: [read]
resource: model-1
comment: user-2 has read access to model-1 only
- permit:
actions: [read]
when:
user_in: resource.namespaces
comment: any user has read access to models with matching attributes
vector_db:
- forbid:
actions: [create, read, delete]
unless:
user_in: role::admin
comment: only user with admin role can use vector_db resources
```
---------
Signed-off-by: Gordon Sim <gsim@redhat.com>
109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
|
# All rights reserved.
|
|
#
|
|
# This source code is licensed under the terms described in the LICENSE file in
|
|
# the root directory of this source tree.
|
|
|
|
from typing import Any
|
|
|
|
from llama_stack.distribution.datatypes import User
|
|
|
|
from .conditions import (
|
|
Condition,
|
|
ProtectedResource,
|
|
parse_conditions,
|
|
)
|
|
from .datatypes import (
|
|
AccessRule,
|
|
Action,
|
|
Scope,
|
|
)
|
|
|
|
|
|
def matches_resource(resource_scope: str, actual_resource: str) -> bool:
|
|
if resource_scope == actual_resource:
|
|
return True
|
|
return resource_scope.endswith("::*") and actual_resource.startswith(resource_scope[:-1])
|
|
|
|
|
|
def matches_scope(
|
|
scope: Scope,
|
|
action: Action,
|
|
resource: str,
|
|
user: str | None,
|
|
) -> bool:
|
|
if scope.resource and not matches_resource(scope.resource, resource):
|
|
return False
|
|
if scope.principal and scope.principal != user:
|
|
return False
|
|
return action in scope.actions
|
|
|
|
|
|
def as_list(obj: Any) -> list[Any]:
|
|
if isinstance(obj, list):
|
|
return obj
|
|
return [obj]
|
|
|
|
|
|
def matches_conditions(
|
|
conditions: list[Condition],
|
|
resource: ProtectedResource,
|
|
user: User,
|
|
) -> bool:
|
|
for condition in conditions:
|
|
# must match all conditions
|
|
if not condition.matches(resource, user):
|
|
return False
|
|
return True
|
|
|
|
|
|
def default_policy() -> list[AccessRule]:
|
|
# for backwards compatibility, if no rules are provided, assume
|
|
# full access subject to previous attribute matching rules
|
|
return [
|
|
AccessRule(
|
|
permit=Scope(actions=list(Action)),
|
|
when=["user in owners " + name for name in ["roles", "teams", "projects", "namespaces"]],
|
|
),
|
|
]
|
|
|
|
|
|
def is_action_allowed(
|
|
policy: list[AccessRule],
|
|
action: Action,
|
|
resource: ProtectedResource,
|
|
user: User | None,
|
|
) -> bool:
|
|
# If user is not set, assume authentication is not enabled
|
|
if not user:
|
|
return True
|
|
|
|
if not len(policy):
|
|
policy = default_policy()
|
|
|
|
qualified_resource_id = resource.type + "::" + resource.identifier
|
|
for rule in policy:
|
|
if rule.forbid and matches_scope(rule.forbid, action, qualified_resource_id, user.principal):
|
|
if rule.when:
|
|
if matches_conditions(parse_conditions(as_list(rule.when)), resource, user):
|
|
return False
|
|
elif rule.unless:
|
|
if not matches_conditions(parse_conditions(as_list(rule.unless)), resource, user):
|
|
return False
|
|
else:
|
|
return False
|
|
elif rule.permit and matches_scope(rule.permit, action, qualified_resource_id, user.principal):
|
|
if rule.when:
|
|
if matches_conditions(parse_conditions(as_list(rule.when)), resource, user):
|
|
return True
|
|
elif rule.unless:
|
|
if not matches_conditions(parse_conditions(as_list(rule.unless)), resource, user):
|
|
return True
|
|
else:
|
|
return True
|
|
# assume access is denied unless we find a rule that permits access
|
|
return False
|
|
|
|
|
|
class AccessDeniedError(RuntimeError):
|
|
pass
|