1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
import asyncio
import os
import uuid
import pytest
from azure.keyvault.administration import KeyVaultDataAction, KeyVaultPermission,KeyVaultRoleScope
from devtools_testutils import add_general_regex_sanitizer, set_bodiless_matcher
from devtools_testutils.aio import recorded_by_proxy_async
from _async_test_case import KeyVaultAccessControlClientPreparer, get_decorator
from _shared.test_case_async import KeyVaultTestCase
from test_access_control import assert_role_definitions_equal
all_api_versions = get_decorator()
class TestAccessControl(KeyVaultTestCase):
def get_replayable_uuid(self, replay_value):
if self.is_live:
value = str(uuid.uuid4())
return value
return replay_value
def get_service_principal_id(self):
replay_value = "service-principal-id"
if self.is_live:
value = os.environ["AZURE_CLIENT_ID"]
return value
return replay_value
@pytest.mark.asyncio
@pytest.mark.parametrize("api_version", all_api_versions)
@KeyVaultAccessControlClientPreparer()
@recorded_by_proxy_async
async def test_role_definitions(self, client, **kwargs):
set_bodiless_matcher()
# list initial role definitions
scope = KeyVaultRoleScope.GLOBAL
original_definitions = []
async for definition in client.list_role_definitions(scope):
original_definitions.append(definition)
assert len(original_definitions)
# create custom role definition
role_name = self.get_resource_name("role-name")
definition_name = self.get_replayable_uuid("definition-name")
add_general_regex_sanitizer(regex=definition_name, value = "definition-name")
permissions = [KeyVaultPermission(data_actions=[KeyVaultDataAction.READ_HSM_KEY])]
created_definition = await client.set_role_definition(
scope=scope,
name=definition_name,
role_name=role_name,
description="test",
permissions=permissions
)
assert "/" in created_definition.assignable_scopes
assert created_definition.role_name == role_name
assert created_definition.name == definition_name
assert created_definition.description == "test"
assert len(created_definition.permissions) == 1
assert created_definition.permissions[0].data_actions == [KeyVaultDataAction.READ_HSM_KEY]
assert created_definition.assignable_scopes == [KeyVaultRoleScope.GLOBAL]
# update custom role definition
permissions = [
KeyVaultPermission(data_actions=[], not_data_actions=[KeyVaultDataAction.READ_HSM_KEY])
]
updated_definition = await client.set_role_definition(
scope=scope, name=definition_name, permissions=permissions
)
assert updated_definition.role_name == ""
assert updated_definition.description == ""
assert len(updated_definition.permissions) == 1
assert len(updated_definition.permissions[0].data_actions) == 0
assert updated_definition.permissions[0].not_data_actions == [KeyVaultDataAction.READ_HSM_KEY]
assert updated_definition.assignable_scopes == [KeyVaultRoleScope.GLOBAL]
# assert that the created role definition isn't duplicated
matching_definitions = []
async for definition in client.list_role_definitions(scope):
if definition.id == updated_definition.id:
matching_definitions.append(definition)
assert len(matching_definitions) == 1
# get custom role definition
definition = await client.get_role_definition(scope=scope, name=definition_name)
assert_role_definitions_equal(definition, updated_definition)
# delete custom role definition
await client.delete_role_definition(scope, definition_name)
async for d in client.list_role_definitions(scope):
assert (d.id != definition.id), "the role definition should have been deleted"
if self.is_live:
await asyncio.sleep(60) # additional waiting to avoid conflicts with resources in other tests
@pytest.mark.asyncio
@pytest.mark.parametrize("api_version", all_api_versions)
@KeyVaultAccessControlClientPreparer()
@recorded_by_proxy_async
async def test_role_assignment(self, client, **kwargs):
set_bodiless_matcher()
scope = KeyVaultRoleScope.GLOBAL
definitions = []
async for definition in client.list_role_definitions(scope):
definitions.append(definition)
# assign an arbitrary role to the service principal authenticating these requests
definition = definitions[0]
principal_id = self.get_service_principal_id()
name = self.get_replayable_uuid("some-uuid")
add_general_regex_sanitizer(regex=name, value = "some-uuid")
created = await client.create_role_assignment(scope, definition.id, principal_id, name=name)
assert created.name == name
#assert created.properties.principal_id == principal_id
assert created.properties.role_definition_id == definition.id
assert created.properties.scope == scope
# should be able to get the new assignment
got = await client.get_role_assignment(scope, name)
assert got.name == name
#assert got.properties.principal_id == principal_id
assert got.properties.role_definition_id == definition.id
assert got.properties.scope == scope
# new assignment should be in the list of all assignments
matching_assignments = []
async for assignment in client.list_role_assignments(scope):
if assignment.role_assignment_id == created.role_assignment_id:
matching_assignments.append(assignment)
assert len(matching_assignments) == 1
# delete the assignment
await client.delete_role_assignment(scope, created.name)
async for assignment in client.list_role_assignments(scope):
assert (
assignment.role_assignment_id != created.role_assignment_id
), "the role assignment should have been deleted"
if self.is_live:
await asyncio.sleep(60) # additional waiting to avoid conflicts with resources in other tests
|
