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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
|
#!/usr/bin/python
# Copyright (C) International Business Machines Corp., 2006
# Author: Stefan Berger <stefanb@us.ibm.com>
# Test to exercise the xspolicy class
from XmTestLib import xapi
from XmTestLib.XenAPIDomain import XmTestAPIDomain
from XmTestLib import *
from xen.xend import XendAPIConstants
import xen.util.xsm.xsm as security
from xen.util import acmpolicy, xsconstants
from xen.util.acmpolicy import ACMPolicy
from xen.xend.XendDomain import DOM0_UUID
from XmTestLib.acm import *
import commands
import os
import base64
if not isACMEnabled():
SKIP("Not running this test since ACM not enabled.")
try:
session = xapi.connect()
except:
SKIP("Skipping this test since xm is not using the Xen-API.")
xm_test = {}
xm_test['policyname'] = "xm-test"
xm_test['date'] = "Fri Sep 29 14:44:38 2006"
xm_test['url'] = None
vm_label_red = "%s:xm-test:red" % xsconstants.ACM_POLICY_ID
vm_label_green = "%s:xm-test:green" % xsconstants.ACM_POLICY_ID
vm_label_blue = "%s:xm-test:blue" % xsconstants.ACM_POLICY_ID
vm_label_sys = "%s:xm-test:SystemManagement" % xsconstants.ACM_POLICY_ID
vm_label_black = "%s:xm-test:black"
session = xapi.connect()
oldlabel = session.xenapi.VM.get_security_label(DOM0_UUID)
ssidref = session.xenapi.VM.set_security_label(DOM0_UUID,
vm_label_sys,
oldlabel)
if int(ssidref) <= 0 or int(ssidref) != 0x00010001:
FAIL("(0) Domain-0 label for '%s' has unexpected failure: %08x" %
(vm_label_sys, int(ssidref)))
print "ssidref for '%s' is 0x%08x" % (vm_label_sys, int(ssidref))
xstype = session.xenapi.XSPolicy.get_xstype()
if int(xstype) & xsconstants.XS_POLICY_ACM == 0:
SKIP("ACM not enabled/compiled in Xen")
policystate = session.xenapi.XSPolicy.get_xspolicy()
if not policystate.has_key('xs_ref'):
FAIL("get_xspolicy must return member 'xs_ref'")
xs_ref = policystate['xs_ref']
if xs_ref != "":
origpolicyxml = session.xenapi.ACMPolicy.get_xml(xs_ref)
else:
origpolicyxml = ""
f = open("xm-test-security_policy.xml", 'r')
if f:
newpolicyxml = f.read()
f.close()
else:
FAIL("Could not read 'xm-test' policy")
try:
os.unlink("/boot/xm-test.bin")
except:
pass
policystate = session.xenapi.XSPolicy.get_xspolicy()
if int(policystate['type']) == 0:
policystate = session.xenapi.XSPolicy.set_xspolicy(
xsconstants.XS_POLICY_ACM,
newpolicyxml,
xsconstants.XS_INST_LOAD | xsconstants.XS_INST_BOOT,
1)
if int(policystate['flags']) == -1:
FAIL("Could not set the new policy.")
print "state of policy = %s " % policystate
rc = session.xenapi.XSPolicy.activate_xspolicy(
policystate['xs_ref'],
xsconstants.XS_INST_LOAD | xsconstants.XS_INST_BOOT)
if int(rc) != xsconstants.XS_INST_LOAD | xsconstants.XS_INST_BOOT:
FAIL("Could not activate the current policy: rc = %08x" % int(rc))
if not os.path.exists("/boot/xm-test.bin"):
FAIL("Binary policy was not installed. Check grub config file.")
policystate = session.xenapi.XSPolicy.get_xspolicy()
if int(policystate['flags']) != xsconstants.XS_INST_BOOT | \
xsconstants.XS_INST_LOAD:
FAIL("Flags (%x) are not indicating the correct state of the policy.",
int(policystate['flags']))
policystate = session.xenapi.XSPolicy.get_xspolicy()
xs_ref = policystate['xs_ref']
newpolicyxml = None
f = open("xm-test-new-security_policy.xml", 'r')
if f:
newpolicyxml = f.read()
f.close()
else:
FAIL("Could not read 'xm-test-new' policy")
cur_acmpol = ACMPolicy(xml = policystate['repr'])
new_acmpol = ACMPolicy(xml = newpolicyxml)
new_acmpol.update_frompolicy(cur_acmpol)
policystate = session.xenapi.XSPolicy.set_xspolicy(xsconstants.XS_POLICY_ACM,
new_acmpol.toxml(),
xsconstants.XS_INST_LOAD | xsconstants.XS_INST_BOOT,
1)
f = open("xm-test-security_policy.xml", 'r')
if f:
newpolicyxml = f.read()
f.close()
else:
FAIL("Could not read 'xm-test-new' policy")
cur_acmpol = new_acmpol
new_acmpol = ACMPolicy(xml = newpolicyxml)
new_acmpol.update_frompolicy(cur_acmpol)
policystate = session.xenapi.XSPolicy.set_xspolicy(xsconstants.XS_POLICY_ACM,
new_acmpol.toxml(),
xsconstants.XS_INST_LOAD | xsconstants.XS_INST_BOOT,
1)
dom0_lab = session.xenapi.VM.get_security_label(DOM0_UUID)
ssidref = session.xenapi.VM.set_security_label(DOM0_UUID,
vm_label_sys, dom0_lab)
if int(ssidref) <= 0 or int(ssidref) != 0x00010001:
FAIL("(1) Domain-0 label for '%s' has unexpected failure: %08x" %
(vm_label_sys, int(ssidref)))
print "ssidref for '%s' is 0x%08x" % (vm_label_sys, int(ssidref))
try:
ssidref = session.xenapi.VM.set_security_label(DOM0_UUID,
vm_label_black,
vm_label_sys)
FAIL("Could set label '%s', although it's not in the policy. "
"ssidref=%s" % (vm_label_black, ssidref))
except:
pass
ssidref = session.xenapi.VM.set_security_label(DOM0_UUID,
vm_label_red,
vm_label_sys)
if int(ssidref) <= 0:
FAIL("(2) Domain-0 label for '%s' has unexpected failure: %08x" %
(vm_label_red, int(ssidref)))
print "ssidref for '%s' is 0x%08x" % (vm_label_red, int(ssidref))
label = session.xenapi.VM.get_security_label(DOM0_UUID)
if label != vm_label_red:
FAIL("Dom0 label '%s' not as expected '%s'" % (label, vm_label_red))
ssidref = session.xenapi.VM.set_security_label(DOM0_UUID,
vm_label_sys,
vm_label_red)
if int(ssidref) <= 0 or int(ssidref) != 0x00010001:
FAIL("(3) Domain-0 label for '%s' has unexpected failure: %08x" %
(vm_label_sys, int(ssidref)))
label = session.xenapi.VM.get_security_label(DOM0_UUID)
if label != vm_label_sys:
FAIL("Dom0 label '%s' not as expected '%s'" % label, dom0_label)
header = session.xenapi.ACMPolicy.get_header(xs_ref)
if header['policyname'] != xm_test['policyname']:
FAIL("Name in header is '%s', expected is '%s'." %
(header['policyname'],xm_test['policyname']))
if header['date'] != xm_test['date']:
FAIL("Date in header is '%s', expected is '%s'." %
(header['date'],xm_test['date']))
if header.has_key("url") and header['url' ] != xm_test['url' ]:
FAIL("URL in header is '%s', expected is '%s'." %
(header['url' ],xm_test['url' ]))
# Create another domain
try:
# XmTestAPIDomain tries to establish a connection to XenD
domain = XmTestAPIDomain(extraConfig={ 'security_label' : vm_label_blue })
except Exception, e:
SKIP("Skipping test. Error: %s" % str(e))
vm_uuid = domain.get_uuid()
res = session.xenapi.VM.get_security_label(vm_uuid)
if res != vm_label_blue:
FAIL("VM has security label '%s', expected is '%s'" %
(res, vm_label_blue))
try:
domain.start(noConsole=True)
except:
FAIL("Could not create domain")
# Attempt to relabel the running domain
ssidref = session.xenapi.VM.set_security_label(vm_uuid,
vm_label_red,
vm_label_blue)
if int(ssidref) <= 0:
FAIL("Could not relabel running domain to '%s'." % vm_label_red)
# user domain is 'red', dom0 is current 'SystemManagement'.
# Try to move domain-0 to 'red' first, then to 'blue'.
# Moving domain-0 to 'red' should work
ssidref = session.xenapi.VM.set_security_label(DOM0_UUID,
vm_label_red,
vm_label_sys)
if int(ssidref) <= 0:
FAIL("Could not label domain-0 '%s'" % vm_label_red)
# Moving the guest domain to 'blue' should not work due to conflict set
try:
ssidref = session.xenapi.VM.set_security_label(vm_uuid,
vm_label_blue,
vm_label_red)
FAIL("Could label guest domain with '%s', although this is in a conflict "
"set. ssidref=%x" % (vm_label_blue,int(ssidref)))
except:
pass
label = session.xenapi.VM.get_security_label(vm_uuid)
if label != vm_label_red:
FAIL("User domain has wrong label '%s', expected '%s'." %
(label, vm_label_red))
label = session.xenapi.VM.get_security_label(DOM0_UUID)
if label != vm_label_red:
FAIL("Domain-0 has wrong label '%s'; expected '%s'." %
(label, vm_label_red))
ssidref = session.xenapi.VM.set_security_label(DOM0_UUID,
vm_label_sys,
vm_label_red)
if int(ssidref) < 0:
FAIL("Could not set the domain-0 security label to '%s'." %
(vm_label_sys))
# pause the domain and relabel it...
session.xenapi.VM.pause(vm_uuid)
label = session.xenapi.VM.get_security_label(vm_uuid)
if label != vm_label_red:
FAIL("User domain has wrong label '%s', expected '%s'." %
(label, vm_label_red))
ssidref = session.xenapi.VM.set_security_label(vm_uuid,
vm_label_blue,
vm_label_red)
print "guest domain new label '%s'; ssidref is 0x%08x" % \
(vm_label_blue, int(ssidref))
if int(ssidref) <= 0:
FAIL("Could not label guest domain with '%s'" % (vm_label_blue))
label = session.xenapi.VM.get_security_label(vm_uuid)
if label != vm_label_blue:
FAIL("User domain has wrong label '%s', expected '%s'." %
(label, vm_label_blue))
session.xenapi.VM.unpause(vm_uuid)
rc = session.xenapi.VM.suspend(vm_uuid)
ssidref = session.xenapi.VM.set_security_label(vm_uuid,
vm_label_green,
vm_label_blue)
print "guest domain new label '%s'; ssidref is 0x%08x" % \
(vm_label_green, int(ssidref))
if int(ssidref) < 0:
FAIL("Could not label suspended guest domain with '%s'" % (vm_label_blue))
label = session.xenapi.VM.get_security_label(vm_uuid)
if label != vm_label_green:
FAIL("User domain has wrong label '%s', expected '%s'." %
(label, vm_label_green))
rc = session.xenapi.VM.resume(vm_uuid, False)
label = session.xenapi.VM.get_security_label(vm_uuid)
if label != vm_label_green:
FAIL("User domain has wrong label '%s', expected '%s'." %
(label, vm_label_green))
|
