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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
|
# pylint: skip-file
# flake8: noqa
# pylint: disable=too-many-instance-attributes
class SecurityContextConstraintsConfig(object):
''' Handle scc options '''
# pylint: disable=too-many-arguments
def __init__(self,
sname,
kubeconfig,
options=None,
fs_group='MustRunAs',
default_add_capabilities=None,
groups=None,
priority=None,
required_drop_capabilities=None,
run_as_user='MustRunAsRange',
se_linux_context='MustRunAs',
supplemental_groups='RunAsAny',
users=None,
annotations=None):
''' constructor for handling scc options '''
self.kubeconfig = kubeconfig
self.name = sname
self.options = options
self.fs_group = fs_group
self.default_add_capabilities = default_add_capabilities
self.groups = groups
self.priority = priority
self.required_drop_capabilities = required_drop_capabilities
self.run_as_user = run_as_user
self.se_linux_context = se_linux_context
self.supplemental_groups = supplemental_groups
self.users = users
self.annotations = annotations
self.data = {}
self.create_dict()
def create_dict(self):
''' assign the correct properties for a scc dict '''
# allow options
if self.options:
for key, value in self.options.items():
self.data[key] = value
else:
self.data['allowHostDirVolumePlugin'] = False
self.data['allowHostIPC'] = False
self.data['allowHostNetwork'] = False
self.data['allowHostPID'] = False
self.data['allowHostPorts'] = False
self.data['allowPrivilegedContainer'] = False
self.data['allowedCapabilities'] = None
# version
self.data['apiVersion'] = 'v1'
# kind
self.data['kind'] = 'SecurityContextConstraints'
# defaultAddCapabilities
self.data['defaultAddCapabilities'] = self.default_add_capabilities
# fsGroup
self.data['fsGroup']['type'] = self.fs_group
# groups
self.data['groups'] = []
if self.groups:
self.data['groups'] = self.groups
# metadata
self.data['metadata'] = {}
self.data['metadata']['name'] = self.name
if self.annotations:
for key, value in self.annotations.items():
self.data['metadata'][key] = value
# priority
self.data['priority'] = self.priority
# requiredDropCapabilities
self.data['requiredDropCapabilities'] = self.required_drop_capabilities
# runAsUser
self.data['runAsUser'] = {'type': self.run_as_user}
# seLinuxContext
self.data['seLinuxContext'] = {'type': self.se_linux_context}
# supplementalGroups
self.data['supplementalGroups'] = {'type': self.supplemental_groups}
# users
self.data['users'] = []
if self.users:
self.data['users'] = self.users
# pylint: disable=too-many-instance-attributes,too-many-public-methods,no-member
class SecurityContextConstraints(Yedit):
''' Class to wrap the oc command line tools '''
default_add_capabilities_path = "defaultAddCapabilities"
fs_group_path = "fsGroup"
groups_path = "groups"
priority_path = "priority"
required_drop_capabilities_path = "requiredDropCapabilities"
run_as_user_path = "runAsUser"
se_linux_context_path = "seLinuxContext"
supplemental_groups_path = "supplementalGroups"
users_path = "users"
kind = 'SecurityContextConstraints'
def __init__(self, content):
'''SecurityContextConstraints constructor'''
super(SecurityContextConstraints, self).__init__(content=content)
self._users = None
self._groups = None
@property
def users(self):
''' users property getter '''
if self._users is None:
self._users = self.get_users()
return self._users
@property
def groups(self):
''' groups property getter '''
if self._groups is None:
self._groups = self.get_groups()
return self._groups
@users.setter
def users(self, data):
''' users property setter'''
self._users = data
@groups.setter
def groups(self, data):
''' groups property setter'''
self._groups = data
def get_users(self):
'''get scc users'''
return self.get(SecurityContextConstraints.users_path) or []
def get_groups(self):
'''get scc groups'''
return self.get(SecurityContextConstraints.groups_path) or []
def add_user(self, inc_user):
''' add a user '''
if self.users:
self.users.append(inc_user)
else:
self.put(SecurityContextConstraints.users_path, [inc_user])
return True
def add_group(self, inc_group):
''' add a group '''
if self.groups:
self.groups.append(inc_group)
else:
self.put(SecurityContextConstraints.groups_path, [inc_group])
return True
def remove_user(self, inc_user):
''' remove a user '''
try:
self.users.remove(inc_user)
except ValueError as _:
return False
return True
def remove_group(self, inc_group):
''' remove a group '''
try:
self.groups.remove(inc_group)
except ValueError as _:
return False
return True
def update_user(self, inc_user):
''' update a user '''
try:
index = self.users.index(inc_user)
except ValueError as _:
return self.add_user(inc_user)
self.users[index] = inc_user
return True
def update_group(self, inc_group):
''' update a group '''
try:
index = self.groups.index(inc_group)
except ValueError as _:
return self.add_group(inc_group)
self.groups[index] = inc_group
return True
def find_user(self, inc_user):
''' find a user '''
index = None
try:
index = self.users.index(inc_user)
except ValueError as _:
return index
return index
def find_group(self, inc_group):
''' find a group '''
index = None
try:
index = self.groups.index(inc_group)
except ValueError as _:
return index
return index
|
