Skip to content

Commit

Permalink
Enforce repo management permission (#4976)
Browse files Browse the repository at this point in the history 
Also fix permission report for kinds relying on global permissions.
  • Loading branch information
@gmazoyer
gmazoyer authored Nov 18, 2024
1 parent 380ea17 commit 0ec2fdc
Show file tree
Hide file tree
Showing 3 changed files with 72 additions and 57 deletions.
2 changes: 2 additions & 0 deletions backend/infrahub/graphql/api/dependencies.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
AccountManagerPermissionChecker,
ObjectPermissionChecker,
PermissionManagerPermissionChecker,
RepositoryManagerPermissionChecker,
)
from ..auth.query_permission_checker.read_only_checker import ReadOnlyGraphQLPermissionChecker
from ..auth.query_permission_checker.read_write_checker import ReadWriteGraphQLPermissionChecker
Expand All @@ -30,6 +31,7 @@ def build_graphql_query_permission_checker() -> GraphQLQueryPermissionChecker:
DefaultBranchPermissionChecker(),
MergeBranchPermissionChecker(),
AccountManagerPermissionChecker(),
RepositoryManagerPermissionChecker(),
PermissionManagerPermissionChecker(),
ObjectPermissionChecker(),
ReadWriteGraphQLPermissionChecker(), # Deprecated, will be replace by either a global permission or object permissions
Expand Down
73 changes: 42 additions & 31 deletions backend/infrahub/permissions/report.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@

from infrahub.core import registry
from infrahub.core.account import GlobalPermission
from infrahub.core.constants import GLOBAL_BRANCH_NAME, GlobalPermissions, PermissionDecision
from infrahub.core.constants import GLOBAL_BRANCH_NAME, GlobalPermissions, InfrahubKind, PermissionDecision
from infrahub.core.schema.node_schema import NodeSchema
from infrahub.permissions.constants import AssignedPermissions, BranchRelativePermissionDecision, PermissionDecisionFlag
from infrahub.permissions.local_backend import LocalPermissionBackend

Expand All @@ -17,28 +18,48 @@
from infrahub.permissions.types import KindPermissions


def get_permission_report(
def get_permission_report( # noqa: PLR0911
backend: PermissionBackend,
permissions: AssignedPermissions,
branch: Branch,
node: MainSchemaTypes,
action: str,
is_super_admin: bool = False,
can_edit_default_branch: bool = False, # pylint: disable=unused-argument
global_permission_report: dict[GlobalPermissions, bool],
) -> BranchRelativePermissionDecision:
is_default_branch = branch.name in (GLOBAL_BRANCH_NAME, registry.default_branch)

if is_super_admin:
if global_permission_report[GlobalPermissions.SUPER_ADMIN]:
return BranchRelativePermissionDecision.ALLOW

if action != "view":
if node.kind in (InfrahubKind.ACCOUNTGROUP, InfrahubKind.ACCOUNTROLE, InfrahubKind.GENERICACCOUNT) or (
isinstance(node, NodeSchema) and InfrahubKind.GENERICACCOUNT in node.inherit_from
):
return (
BranchRelativePermissionDecision.ALLOW
if global_permission_report[GlobalPermissions.MANAGE_ACCOUNTS]
else BranchRelativePermissionDecision.DENY
)
if node.kind in (InfrahubKind.BASEPERMISSION, InfrahubKind.GLOBALPERMISSION, InfrahubKind.OBJECTPERMISSION) or (
isinstance(node, NodeSchema) and InfrahubKind.BASEPERMISSION in node.inherit_from
):
return (
BranchRelativePermissionDecision.ALLOW
if global_permission_report[GlobalPermissions.MANAGE_PERMISSIONS]
else BranchRelativePermissionDecision.DENY
)
if node.kind in (InfrahubKind.GENERICREPOSITORY, InfrahubKind.REPOSITORY, InfrahubKind.READONLYREPOSITORY) or (
isinstance(node, NodeSchema) and InfrahubKind.GENERICREPOSITORY in node.inherit_from
):
return (
BranchRelativePermissionDecision.ALLOW
if global_permission_report[GlobalPermissions.MANAGE_REPOSITORIES]
else BranchRelativePermissionDecision.DENY
)

is_default_branch = branch.name in (GLOBAL_BRANCH_NAME, registry.default_branch)
decision = backend.report_object_permission(
permissions=permissions["object_permissions"], namespace=node.namespace, name=node.name, action=action
)

# What do we do if edit default branch global permission is set?
# if can_edit_default_branch:
# decision |= PermissionDecisionFlag.ALLOW_DEFAULT

if (
decision == PermissionDecisionFlag.ALLOW_ALL
or (decision & PermissionDecisionFlag.ALLOW_DEFAULT and is_default_branch)
Expand All @@ -59,18 +80,12 @@ async def report_schema_permissions(
perm_backend = LocalPermissionBackend()
permissions = await perm_backend.load_permissions(db=db, account_session=account_session, branch=branch)

is_super_admin = perm_backend.resolve_global_permission(
permissions=permissions["global_permissions"],
permission_to_check=GlobalPermission(
action=GlobalPermissions.SUPER_ADMIN.value, decision=PermissionDecision.ALLOW_ALL.value
),
)
can_edit_default_branch = perm_backend.resolve_global_permission(
permissions=permissions["global_permissions"],
permission_to_check=GlobalPermission(
action=GlobalPermissions.EDIT_DEFAULT_BRANCH.value, decision=PermissionDecision.ALLOW_ALL.value
),
)
global_permission_report: dict[GlobalPermissions, bool] = {}
for perm in GlobalPermissions:
global_permission_report[perm] = perm_backend.resolve_global_permission(
permissions=permissions["global_permissions"],
permission_to_check=GlobalPermission(action=perm.value, decision=PermissionDecision.ALLOW_ALL.value),
)

permission_objects: list[KindPermissions] = []
for node in schemas:
Expand All @@ -83,35 +98,31 @@ async def report_schema_permissions(
branch=branch,
node=node,
action="create",
is_super_admin=is_super_admin,
can_edit_default_branch=can_edit_default_branch,
global_permission_report=global_permission_report,
),
"delete": get_permission_report(
backend=perm_backend,
permissions=permissions,
branch=branch,
node=node,
action="delete",
is_super_admin=is_super_admin,
can_edit_default_branch=can_edit_default_branch,
global_permission_report=global_permission_report,
),
"update": get_permission_report(
backend=perm_backend,
permissions=permissions,
branch=branch,
node=node,
action="update",
is_super_admin=is_super_admin,
can_edit_default_branch=can_edit_default_branch,
global_permission_report=global_permission_report,
),
"view": get_permission_report(
backend=perm_backend,
permissions=permissions,
branch=branch,
node=node,
action="view",
is_super_admin=is_super_admin,
can_edit_default_branch=can_edit_default_branch,
global_permission_report=global_permission_report,
),
}
)
Expand Down
54 changes: 28 additions & 26 deletions backend/tests/unit/graphql/queries/test_list_permissions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,9 @@
}
"""

REPOSITORY_QUERY = """
IPAM_IP_NAMESPACE_QUERY = """
query {
CoreGenericRepository {
BuiltinIPNamespace {
permissions {
count
edges {
Expand All @@ -62,9 +62,9 @@
"""


QUERY_ACCOUNT_ROLE = """
QUERY_IP_PREFIX_POOL = """
query {
CoreAccountRole {
CoreIPPrefixPool {
edges {
node {
display_label
Expand Down Expand Up @@ -132,6 +132,18 @@ async def test_setup(
action=PermissionAction.VIEW.value,
decision=PermissionDecisionFlag.ALLOW_ALL,
),
ObjectPermission(
namespace="Ipam",
name="*",
action=PermissionAction.ANY.value,
decision=PermissionDecisionFlag.ALLOW_OTHER,
),
ObjectPermission(
namespace="Ipam",
name="*",
action=PermissionAction.VIEW.value,
decision=PermissionDecisionFlag.ALLOW_ALL,
),
]:
obj = await Node.init(db=db, schema=InfrahubKind.OBJECTPERMISSION)
await obj.new(
Expand Down Expand Up @@ -215,42 +227,33 @@ async def test_first_account_list_permissions_for_generics(

result = await graphql(
schema=gql_params.schema,
source=REPOSITORY_QUERY,
source=IPAM_IP_NAMESPACE_QUERY,
context_value=gql_params.context,
)

assert not result.errors
assert result.data
assert result.data["CoreGenericRepository"]["permissions"]["count"] == 3
assert result.data["BuiltinIPNamespace"]["permissions"]["count"] == 2
assert {
"node": {
"kind": "CoreGenericRepository",
"kind": "BuiltinIPNamespace",
"create": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"update": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"delete": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"view": BranchRelativePermissionDecision.ALLOW.name,
}
} in result.data["CoreGenericRepository"]["permissions"]["edges"]
assert {
"node": {
"kind": "CoreRepository",
"create": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"update": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"update": BranchRelativePermissionDecision.DENY.name,
"delete": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"view": BranchRelativePermissionDecision.ALLOW.name,
}
} in result.data["CoreGenericRepository"]["permissions"]["edges"]
} in result.data["BuiltinIPNamespace"]["permissions"]["edges"]
assert {
"node": {
"kind": "CoreReadOnlyRepository",
"kind": "IpamNamespace",
"create": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"update": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"delete": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"view": BranchRelativePermissionDecision.ALLOW.name,
}
} in result.data["CoreGenericRepository"]["permissions"]["edges"]
} in result.data["BuiltinIPNamespace"]["permissions"]["edges"]

async def test_first_account_account_role(
async def test_first_account_ipprefix_pool(
self, db: InfrahubDatabase, permissions_helper: PermissionsHelper
) -> None:
"""In the main branch the first account doesn't have the permission to make changes, but it has in the other branches"""
Expand All @@ -261,21 +264,20 @@ async def test_first_account_account_role(
db=db, include_mutation=True, branch=permissions_helper.default_branch, account_session=session
)

result = await graphql(schema=gql_params.schema, source=QUERY_ACCOUNT_ROLE, context_value=gql_params.context)
result = await graphql(schema=gql_params.schema, source=QUERY_IP_PREFIX_POOL, context_value=gql_params.context)

assert not result.errors
assert result.data
assert result.data["CoreAccountRole"]["permissions"]["count"] == 1
assert result.data["CoreAccountRole"]["permissions"]["edges"][0] == {
assert result.data["CoreIPPrefixPool"]["permissions"]["count"] == 1
assert result.data["CoreIPPrefixPool"]["permissions"]["edges"][0] == {
"node": {
"kind": "CoreAccountRole",
"kind": "CoreIPPrefixPool",
"create": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"update": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"delete": BranchRelativePermissionDecision.ALLOW_OTHER.name,
"view": BranchRelativePermissionDecision.ALLOW.name,
}
}
assert result.data["CoreAccountRole"]["edges"][0]["node"]["display_label"] == "admin"


QUERY_TAGS_ATTR = """
Expand Down

0 comments on commit 0ec2fdc

Please sign in to comment.