xen-project/xen
1
0 Fork 1
mirror of https://github.com/xen-project/xen.git synced 2026-10-28 18:21:39 +01:00
Code Issues Activity

This patch adds new python access control management scripts, which

Browse Source 
integrate into Xen Management and which support the new access control 
labels (labels replace the ssidref numbers at the management user 
interface).

Signed-off by: Reiner Sailer <sailer@us.ibm.com>
This commit is contained in:
smh22@firebug.cl.cam.ac.uk committed 2006-04-24 10:58:25 +01:00
1 parent 2d59e34fc2
commit e7f4f3a1de
11 files changed
+1354 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tools/python/setup.py
+8 -1
View File
@@ -31,6 +31,13 @@ xs = Extension("xs",
libraries = libraries,
sources = [ "xen/lowlevel/xs/xs.c" ])
acm = Extension("acm",
extra_compile_args = extra_compile_args,
include_dirs = include_dirs + [ "xen/lowlevel/acm" ],
library_dirs = library_dirs,
libraries = libraries,
sources = [ "xen/lowlevel/acm/acm.c" ])
setup(name = 'xen',
version = '3.0',
description = 'Xen',
@@ -50,7 +57,7 @@ setup(name = 'xen',
'xen.xm.tests'
],
ext_package = "xen.lowlevel",
ext_modules = [ xc, xs ]
ext_modules = [ xc, xs, acm ]
)
os.chdir('logging')
tools/python/xen/lowlevel/acm/acm.c
+237
View File
@@ -0,0 +1,237 @@
/****************************************************************
* acm.c
*
* Copyright (C) 2006 IBM Corporation
*
* Authors:
* Reiner Sailer <sailer@watson.ibm.com>
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License as
* published by the Free Software Foundation, version 2 of the
* License.
*
* ACM low-level code that allows Python control code to leverage
* the ACM hypercall interface to retrieve real-time information
* from the Xen hypervisor security module.
*
* indent -i4 -kr -nut
*/
#include <Python.h>
#include <stdio.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/types.h>
#include <stdlib.h>
#include <sys/ioctl.h>
#include <netinet/in.h>
#include <xen/acm.h>
#include <xen/acm_ops.h>
#include <xen/linux/privcmd.h>
#define PERROR(_m, _a...) \
fprintf(stderr, "ERROR: " _m " (%d = %s)\n" , ## _a , \
errno, strerror(errno))
static inline int do_acm_op(int xc_handle, struct acm_op *op)
{
int ret = -1;
privcmd_hypercall_t hypercall;
op->interface_version = ACM_INTERFACE_VERSION;
hypercall.op = __HYPERVISOR_acm_op;
hypercall.arg[0] = (unsigned long) op;
if (mlock(op, sizeof(*op)) != 0) {
PERROR("Could not lock memory for Xen policy hypercall");
goto out1;
}
ret = ioctl(xc_handle, IOCTL_PRIVCMD_HYPERCALL, &hypercall);
if (ret < 0) {
if (errno == EACCES)
PERROR("ACM operation failed.");
goto out2;
}
out2:
munlock(op, sizeof(*op));
out1:
return ret;
}
/* generic shared function */
void * __getssid(int domid, uint32_t *buflen)
{
struct acm_op op;
int acm_cmd_fd;
#define SSID_BUFFER_SIZE 4096
void *buf = NULL;
if ((acm_cmd_fd = open("/proc/xen/privcmd", O_RDONLY)) < 0) {
goto out1;
}
if ((buf = malloc(SSID_BUFFER_SIZE)) == NULL) {
PERROR("acm.policytype: Could not allocate ssid buffer!\n");
goto out2;
}
memset(buf, 0, SSID_BUFFER_SIZE);
op.cmd = ACM_GETSSID;
op.interface_version = ACM_INTERFACE_VERSION;
op.u.getssid.ssidbuf = buf;
op.u.getssid.ssidbuf_size = SSID_BUFFER_SIZE;
op.u.getssid.get_ssid_by = DOMAINID;
op.u.getssid.id.domainid = domid;
if (do_acm_op(acm_cmd_fd, &op) < 0) {
free(buf);
buf = NULL;
goto out2;
} else {
*buflen = SSID_BUFFER_SIZE;
goto out2;
}
out2:
close(acm_cmd_fd);
out1:
return buf;
}
/* retrieve the policytype indirectly by retrieving the
* ssidref for domain 0 (always exists) */
static PyObject *policy(PyObject * self, PyObject * args)
{
/* out */
char *policyreference;
PyObject *ret = NULL;
void *ssid_buffer;
uint32_t buf_len;
if (!PyArg_ParseTuple(args, "", NULL)) {
goto out1;
}
ssid_buffer = __getssid(0, &buf_len);
if (ssid_buffer == NULL) {
goto out1;
} else if (buf_len < sizeof(struct acm_ssid_buffer)) {
goto out2;
} else {
struct acm_ssid_buffer *ssid = (struct acm_ssid_buffer *)ssid_buffer;
policyreference = (char *)(ssid_buffer + ssid->policy_reference_offset
+ sizeof (struct acm_policy_reference_buffer));
}
ret = Py_BuildValue("s", policyreference);
out2:
free(ssid_buffer);
out1:
return ret;
}
/* retrieve ssid info for a domain domid*/
static PyObject *getssid(PyObject * self, PyObject * args)
{
/* in */
uint32_t domid;
/* out */
char *policytype, *policyreference;
uint32_t ssidref;
void *ssid_buffer;
uint32_t buf_len;
if (!PyArg_ParseTuple(args, "i", &domid)) {
return NULL;
}
ssid_buffer = __getssid(domid, &buf_len);
if (ssid_buffer == NULL) {
return NULL;
} else if (buf_len < sizeof(struct acm_ssid_buffer)) {
free(ssid_buffer);
return NULL;
} else {
struct acm_ssid_buffer *ssid = (struct acm_ssid_buffer *) ssid_buffer;
policytype = ACM_POLICY_NAME(ssid->secondary_policy_code << 4 |
ssid->primary_policy_code);
ssidref = ssid->ssidref;
policyreference = (char *)(ssid_buffer + ssid->policy_reference_offset
+ sizeof (struct acm_policy_reference_buffer));
}
free(ssid_buffer);
return Py_BuildValue("{s:s,s:s,s:i}",
"policyreference", policyreference,
"policytype", policytype,
"ssidref", ssidref);
}
/* retrieve access decision based on domain ids or ssidrefs */
static PyObject *getdecision(PyObject * self, PyObject * args)
{
char *arg1_name, *arg1, *arg2_name, *arg2, *decision = NULL;
struct acm_op op;
int acm_cmd_fd, ret;
if (!PyArg_ParseTuple(args, "ssss", &arg1_name, &arg1, &arg2_name, &arg2)) {
return NULL;
}
if ((acm_cmd_fd = open("/proc/xen/privcmd", O_RDONLY)) <= 0) {
PERROR("Could not open xen privcmd device!\n");
return NULL;
}
if ((strcmp(arg1_name, "domid") && strcmp(arg1_name, "ssidref")) ||
(strcmp(arg2_name, "domid") && strcmp(arg2_name, "ssidref")))
return NULL;
op.cmd = ACM_GETDECISION;
op.interface_version = ACM_INTERFACE_VERSION;
op.u.getdecision.hook = SHARING;
if (!strcmp(arg1_name, "domid")) {
op.u.getdecision.get_decision_by1 = DOMAINID;
op.u.getdecision.id1.domainid = atoi(arg1);
} else {
op.u.getdecision.get_decision_by1 = SSIDREF;
op.u.getdecision.id1.ssidref = atol(arg1);
}
if (!strcmp(arg2_name, "domid")) {
op.u.getdecision.get_decision_by2 = DOMAINID;
op.u.getdecision.id2.domainid = atoi(arg2);
} else {
op.u.getdecision.get_decision_by2 = SSIDREF;
op.u.getdecision.id2.ssidref = atol(arg2);
}
ret = do_acm_op(acm_cmd_fd, &op);
close(acm_cmd_fd);
if (op.u.getdecision.acm_decision == ACM_ACCESS_PERMITTED)
decision = "PERMITTED";
else if (op.u.getdecision.acm_decision == ACM_ACCESS_DENIED)
decision = "DENIED";
return Py_BuildValue("s", decision);
}
/*=================General Python Extension Declarations=================*/
/* methods */
static PyMethodDef acmMethods[] = {
{"policy", policy, METH_VARARGS, "Retrieve Active ACM Policy Reference Name"},
{"getssid", getssid, METH_VARARGS, "Retrieve label information and ssidref for a domain"},
{"getdecision", getdecision, METH_VARARGS, "Retrieve ACM access control decision"},
/* end of list (extend list above this line) */
{NULL, NULL, 0, NULL}
};
/* inits */
PyMODINIT_FUNC initacm(void)
{
Py_InitModule("acm", acmMethods);
}
tools/python/xen/util/security.py
+504
View File
@@ -0,0 +1,504 @@
#===========================================================================
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#============================================================================
# Copyright (C) 2006 International Business Machines Corp.
# Author: Reiner Sailer
#============================================================================
import commands
import logging
import sys, os, string, re
import traceback
import shutil
from xen.lowlevel import acm
from xen.xend import sxp
#global directories and tools for security management
policy_dir_prefix = "/etc/xen/acm-security/policies"
boot_filename = "/boot/grub/menu.lst"
xensec_xml2bin = "/usr/sbin/xensec_xml2bin"
xensec_tool = "/usr/sbin/xensec_tool"
#global patterns for map file
#police_reference_tagname = "POLICYREFERENCENAME"
primary_entry_re = re.compile("\s*PRIMARY\s+.*", re.IGNORECASE)
secondary_entry_re = re.compile("\s*SECONDARY\s+.*", re.IGNORECASE)
label_template_re = re.compile(".*security_label_template.xml", re.IGNORECASE)
mapping_filename_re = re.compile(".*\.map", re.IGNORECASE)
policy_reference_entry_re = re.compile("\s*POLICYREFERENCENAME\s+.*", re.IGNORECASE)
vm_label_re = re.compile("\s*LABEL->SSID\s+VM\s+.*", re.IGNORECASE)
res_label_re = re.compile("\s*LABEL->SSID\s+RES\s+.*", re.IGNORECASE)
all_label_re = re.compile("\s*LABEL->SSID\s+.*", re.IGNORECASE)
access_control_re = re.compile("\s*access_control\s*=", re.IGNORECASE)
#global patterns for boot configuration file
xen_title_re = re.compile("\s*title\s+XEN", re.IGNORECASE)
any_title_re = re.compile("\s*title\s", re.IGNORECASE)
xen_kernel_re = re.compile("\s*kernel.*xen.*\.gz", re.IGNORECASE)
kernel_ver_re = re.compile("\s*module.*vmlinuz", re.IGNORECASE)
any_module_re = re.compile("\s*module\s", re.IGNORECASE)
empty_line_re = re.compile("^\s*$")
binary_name_re = re.compile(".*[chwall|ste|chwall_ste].*\.bin", re.IGNORECASE)
policy_name_re = re.compile(".*[chwall|ste|chwall_ste].*", re.IGNORECASE)
log = logging.getLogger("xend.util.security")
# Our own exception definition. It is masked (pass) if raised and
# whoever raises this exception must provide error information.
class ACMError(Exception):
def __init__(self,value):
self.value = value
def __str__(self):
return repr(self.value)
def err(msg):
"""Raise ACM exception.
"""
sys.stderr.write("ACMError: " + msg + "\n")
raise ACMError(msg)
active_policy = None
def refresh_security_policy():
"""
retrieves security policy
"""
global active_policy
try:
active_policy = acm.policy()
except:
active_policy = "INACTIVE"
# now set active_policy
refresh_security_policy()
def on():
"""
returns none if security policy is off (not compiled),
any string otherwise, use it: if not security.on() ...
"""
refresh_security_policy()
return (active_policy not in ['INACTIVE', 'NULL'])
# Assumes a 'security' info [security access_control ...] [ssidref ...]
def get_security_info(info, field):
"""retrieves security field from self.info['security'])
allowed search fields: ssidref, label, policy
"""
if isinstance(info, dict):
security = info['security']
elif isinstance(info, list):
security = sxp.child_value(info, 'security', )
if not security:
if field == 'ssidref':
#return default ssid
return 0
else:
err("Security information not found in info struct.")
if field == 'ssidref':
search = 'ssidref'
elif field in ['policy', 'label']:
search = 'access_control'
else:
err("Illegal field in get_security_info.")
for idx in range(0, len(security)):
if search != security[idx][0]:
continue
if search == 'ssidref':
return int(security[idx][1])
else:
for aidx in range(0, len(security[idx])):
if security[idx][aidx][0] == field:
return str(security[idx][aidx][1])
if search == 'ssidref':
return 0
else:
return None
def get_security_printlabel(info):
"""retrieves printable security label from self.info['security']),
preferably the label name and otherwise (if label is not specified
in config and cannot be found in mapping file) a hex string of the
ssidref or none if both not available
"""
try:
if not on():
return "INACTIVE"
if active_policy in ["DEFAULT"]:
return "DEFAULT"
printlabel = get_security_info(info, 'label')
if printlabel:
return printlabel
ssidref = get_security_info(info, 'ssidref')
if not ssidref:
return None
#try to translate ssidref to a label
result = ssidref2label(ssidref)
if not result:
printlabel = "0x%08x" % ssidref
else:
printlabel = result
return printlabel
except ACMError:
#don't throw an exception in xm list
return "ERROR"
def getmapfile(policyname):
"""
in: if policyname is None then the currently
active hypervisor policy is used
out: 1. primary policy, 2. secondary policy,
3. open file descriptor for mapping file, and
4. True if policy file is available, False otherwise
"""
if not policyname:
policyname = active_policy
map_file_ok = False
primary = None
secondary = None
#strip last part of policy as file name part
policy_dir_list = string.split(policyname, ".")
policy_file = policy_dir_list.pop()
if len(policy_dir_list) > 0:
policy_dir = string.join(policy_dir_list, "/") + "/"
else:
policy_dir = ""
map_filename = policy_dir_prefix + "/" + policy_dir + policy_file + ".map"
# check if it is there, if not check if policy file is there
if not os.path.isfile(map_filename):
policy_filename = policy_dir_prefix + "/" + policy_dir + policy_file + "-security_policy.xml"
if not os.path.isfile(policy_filename):
err("Policy file \'" + policy_filename + "\' not found.")
else:
err("Mapping file \'" + map_filename + "\' not found." +
" Use xm makepolicy to create it.")
f = open(map_filename)
for line in f:
if policy_reference_entry_re.match(line):
l = line.split()
if (len(l) == 2) and (l[1] == policyname):
map_file_ok = True
elif primary_entry_re.match(line):
l = line.split()
if len(l) == 2:
primary = l[1]
elif secondary_entry_re.match(line):
l = line.split()
if len(l) == 2:
secondary = l[1]
f.close()
f = open(map_filename)
if map_file_ok and primary and secondary:
return (primary, secondary, f, True)
else:
err("Mapping file inconsistencies found. Try makepolicy to create a new one.")
def ssidref2label(ssidref_var):
"""
returns labelname corresponding to ssidref;
maps current policy to default directory
to find mapping file
"""
#1. translated permitted input formats
if isinstance(ssidref_var, str):
ssidref_var.strip()
if ssidref_var[0:2] == "0x":
ssidref = int(ssidref_var[2:], 16)
else:
ssidref = int(ssidref_var)
elif isinstance(ssidref_var, int):
ssidref = ssidref_var
else:
err("Instance type of ssidref not supported (must be of type 'str' or 'int')")
(primary, secondary, f, pol_exists) = getmapfile(None)
if not f:
if (pol_exists):
err("Mapping file for policy \'" + policyname + "\' not found.\n" +
"Please use makepolicy command to create mapping file!")
else:
err("Policy file for \'" + active_policy + "\' not found.")
#2. get labelnames for both ssidref parts
pri_ssid = ssidref & 0xffff
sec_ssid = ssidref >> 16
pri_labels = []
sec_labels = []
labels = []
for line in f:
l = line.split()
if (len(l) < 5) or (l[0] != "LABEL->SSID"):
continue
if primary and (l[2] == primary) and (int(l[4], 16) == pri_ssid):
pri_labels.append(l[3])
if secondary and (l[2] == secondary) and (int(l[4], 16) == sec_ssid):
sec_labels.append(l[3])
f.close()
#3. get the label that is in both lists (combination must be a single label)
if secondary == "NULL":
labels = pri_labels
else:
for i in pri_labels:
for j in sec_labels:
if (i==j):
labels.append(i)
if len(labels) != 1:
err("Label for ssidref \'" + str(ssidref) +
"\' unknown or not unique in policy \'" + active_policy + "\'")
return labels[0]
def label2ssidref(labelname, policyname):
"""
returns ssidref corresponding to labelname;
maps current policy to default directory
to find mapping file """
if policyname in ['NULL', 'INACTIVE', 'DEFAULT']:
err("Cannot translate labels for \'" + policyname + "\' policy.")
(primary, secondary, f, pol_exists) = getmapfile(policyname)
#2. get labelnames for ssidref parts and find a common label
pri_ssid = []
sec_ssid = []
for line in f:
l = line.split()
if (len(l) < 5) or (l[0] != "LABEL->SSID"):
continue
if primary and (l[2] == primary) and (l[3] == labelname):
pri_ssid.append(int(l[4], 16))
if secondary and (l[2] == secondary) and (l[3] == labelname):
sec_ssid.append(int(l[4], 16))
f.close()
#3. sanity check and composition of ssidref
if (len(pri_ssid) == 0) or ((len(sec_ssid) == 0) and (secondary != "NULL")):
err("Label \'" + labelname + "\' not found.")
elif (len(pri_ssid) > 1) or (len(sec_ssid) > 1):
err("Label \'" + labelname + "\' not unique in policy (policy error)")
if secondary == "NULL":
return pri_ssid[0]
else:
return (sec_ssid[0] << 16) | pri_ssid[0]
def refresh_ssidref(config):
"""
looks up ssidref from security field
and refreshes the value if label exists
"""
#called by dom0, policy could have changed after xen.utils.security was initialized
refresh_security_policy()
security = None
if isinstance(config, dict):
security = config['security']
elif isinstance(config, list):
security = sxp.child_value(config, 'security',)
else:
err("Instance type of config parameter not supported.")
if not security:
#nothing to do (no security label attached)
return config
policyname = None
labelname = None
# compose new security field
for idx in range(0, len(security)):
if security[idx][0] == 'ssidref':
security.pop(idx)
break
elif security[idx][0] == 'access_control':
for jdx in [1, 2]:
if security[idx][jdx][0] == 'label':
labelname = security[idx][jdx][1]
elif security[idx][jdx][0] == 'policy':
policyname = security[idx][jdx][1]
else:
err("Illegal field in access_control")
#verify policy is correct
if active_policy != policyname:
err("Policy \'" + policyname + "\' in label does not match active policy \'"
+ active_policy +"\'!")
new_ssidref = label2ssidref(labelname, policyname)
if not new_ssidref:
err("SSIDREF refresh failed!")
security.append([ 'ssidref',str(new_ssidref)])
security = ['security', security ]
for idx in range(0,len(config)):
if config[idx][0] == 'security':
config.pop(idx)
break
config.append(security)
def get_ssid(domain):
"""
enables domains to retrieve the label / ssidref of a running domain
"""
if not on():
err("No policy active.")
if isinstance(domain, str):
domain_int = int(domain)
elif isinstance(domain, int):
domain_int = domain
else:
err("Illegal parameter type.")
try:
ssid_info = acm.getssid(int(domain_int))
except:
err("Cannot determine security information.")
if active_policy in ["DEFAULT"]:
label = "DEFAULT"
else:
label = ssidref2label(ssid_info["ssidref"])
return(ssid_info["policyreference"],
label,
ssid_info["policytype"],
ssid_info["ssidref"])
def get_decision(arg1, arg2):
"""
enables domains to retrieve access control decisions from
the hypervisor Access Control Module.
IN: args format = ['domid', id] or ['ssidref', ssidref]
or ['access_control', ['policy', policy], ['label', label]]
"""
if not on():
err("No policy active.")
#translate labels before calling low-level function
if arg1[0] == 'access_control':
if (arg1[1][0] != 'policy') or (arg1[2][0] != 'label') :
err("Argument type not supported.")
ssidref = label2ssidref(arg1[2][1], arg1[1][1])
arg1 = ['ssidref', str(ssidref)]
if arg2[0] == 'access_control':
if (arg2[1][0] != 'policy') or (arg2[2][0] != 'label') :
err("Argument type not supported.")
ssidref = label2ssidref(arg2[2][1], arg2[1][1])
arg2 = ['ssidref', str(ssidref)]
try:
decision = acm.getdecision(arg1[0], arg1[1], arg2[0], arg2[1])
except:
err("Cannot determine decision.")
if decision:
return decision
else:
err("Cannot determine decision (Invalid parameter).")
def make_policy(policy_name):
policy_file = string.join(string.split(policy_name, "."), "/")
if not os.path.isfile(policy_dir_prefix + "/" + policy_file + "-security_policy.xml"):
err("Unknown policy \'" + policy_name + "\'")
(ret, output) = commands.getstatusoutput(xensec_xml2bin + " -d " + policy_dir_prefix + " " + policy_file)
if ret:
err("Creating policy failed:\n" + output)
def load_policy(policy_name):
global active_policy
policy_file = policy_dir_prefix + "/" + string.join(string.split(policy_name, "."), "/")
if not os.path.isfile(policy_file + ".bin"):
if os.path.isfile(policy_file + "-security_policy.xml"):
err("Binary file does not exist." +
"Please use makepolicy to build the policy binary.")
else:
err("Unknown Policy " + policy_name)
#require this policy to be the first or the same as installed
if active_policy not in ['DEFAULT', policy_name]:
err("Active policy \'" + active_policy +
"\' incompatible with new policy \'" + policy_name + "\'")
(ret, output) = commands.getstatusoutput(xensec_tool + " loadpolicy " + policy_file + ".bin")
if ret:
err("Loading policy failed:\n" + output)
else:
# refresh active policy
refresh_security_policy()
def dump_policy():
if active_policy in ['NULL', 'INACTIVE']:
err("\'" + active_policy + "\' policy. Nothing to dump.")
(ret, output) = commands.getstatusoutput(xensec_tool + " getpolicy")
if ret:
err("Dumping hypervisor policy failed:\n" + output)
print output
def list_labels(policy_name, condition):
if (not policy_name) and (active_policy) in ["NULL", "INACTIVE", "DEFAULT"]:
err("Current policy \'" + active_policy + "\' has no labels defined.\n")
(primary, secondary, f, pol_exists) = getmapfile(policy_name)
if not f:
if pol_exists:
err("Cannot find mapfile for policy \'" + policy_name +
"\'.\nPlease use makepolicy to create mapping file.")
else:
err("Unknown policy \'" + policy_name + "\'")
labels = []
for line in f:
if condition.match(line):
label = line.split()[3]
if label not in labels:
labels.append(label)
return labels
tools/python/xen/xm/addlabel.py
+76
View File
@@ -0,0 +1,76 @@
#============================================================================
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#============================================================================
# Copyright (C) 2006 International Business Machines Corp.
# Author: Reiner Sailer <sailer@us.ibm.com>
#============================================================================
"""Labeling a domain configuration file.
"""
import sys, os
import traceback
from xen.util.security import ACMError, err, active_policy, label2ssidref, on, access_control_re
def usage():
print "\nUsage: xm addlabel <configfile> <label> [<policy>]\n"
print " This program adds an acm_label entry into the 'configfile'."
print " It derives the policy from the running hypervisor if it"
print " is not given (optional parameter). If the configfile is"
print " already labeled, then addlabel fails.\n"
err("Usage")
def main(argv):
try:
policyref = None
if len(argv) not in [3,4]:
usage()
configfile = argv[1]
label = argv[2]
if len(argv) == 4:
policyref = argv[3]
elif on():
policyref = active_policy
else:
err("No active policy. Policy must be specified in command line.")
#sanity checks: make sure this label can be instantiated later on
ssidref = label2ssidref(label, policyref)
new_label = "access_control = ['policy=%s,label=%s']\n" % (policyref, label)
if not os.path.isfile(configfile):
err("Configuration file \'" + configfile + "\' not found.")
config_fd = open(configfile, "ra+")
for line in config_fd:
if not access_control_re.match(line):
continue
config_fd.close()
err("Config file \'" + configfile + "\' is already labeled.")
config_fd.write(new_label)
config_fd.close()
except ACMError:
pass
except:
traceback.print_exc(limit=1)
if __name__ == '__main__':
main(sys.argv)
tools/python/xen/xm/cfgbootpolicy.py
+188
View File
@@ -0,0 +1,188 @@
#============================================================================
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#============================================================================
# Copyright (C) 2006 International Business Machines Corp.
# Author: Reiner Sailer <sailer@us.ibm.com>
#============================================================================
"""Configuring a security policy into the boot configuration
"""
import sys
import traceback
import tempfile
import os, stat
import re
import commands
import shutil
import string
from xen.util.security import ACMError, err
from xen.util.security import policy_dir_prefix, boot_filename, xen_title_re
from xen.util.security import any_title_re, xen_kernel_re, kernel_ver_re, any_module_re
from xen.util.security import empty_line_re, binary_name_re, policy_name_re
def usage():
print "\nUsage: xm cfgbootpolicy <policy> [<kernelversion>]\n"
print " Adds a 'module' line to the Xen grub.conf entry"
print " so that xen boots into a specific access control"
print " policy. If kernelversion is not given, then this"
print " script tries to determine it by looking for a grub"
print " entry with a line kernel xen.* If there are multiple"
print " Xen entries, then it must be called with an explicit"
print " version (it will fail otherwise).\n"
err("Usage")
def determine_kernelversion(user_specified):
within_xen_title = 0
within_xen_entry = 0
version_list = []
guess_version = None
grub_fd = open(boot_filename)
for line in grub_fd:
if xen_title_re.match(line):
within_xen_title = 1
elif within_xen_title and xen_kernel_re.match(line):
within_xen_entry = 1
elif within_xen_title and within_xen_entry and kernel_ver_re.match(line):
for i in line.split():
if (i.find("vmlinuz-") >= 0):
# skip start until "vmlinuz-"
guess_version = i[i.find("vmlinuz-") + len("vmlinuz-"):]
if user_specified:
if (guess_version == user_specified):
version_list.append(guess_version)
else:
version_list.append(guess_version)
elif len(line.split()) > 0:
if line.split()[0] == "title":
within_xen_title = 0
within_xen_entry = 0
if len(version_list) > 1:
err("Cannot decide between entries for kernels: " + version_list)
elif len(version_list) == 0:
err("Cannot find a boot entry candidate (please create a Xen boot entry first).")
else:
return version_list[0]
def insert_policy(boot_file, kernel_version, policy_name):
"""
inserts policy binary file as last line of the grub entry
matching the kernel_version version
"""
within_xen_title = 0
within_xen_entry = 0
insert_at_end_of_entry = 0
path_prefix = ''
done = False
(tmp_fd, tmp_grub) = tempfile.mkstemp()
#follow symlink since menue.lst might be linked to grub.conf
if stat.S_ISLNK(os.lstat(boot_file)[stat.ST_MODE]):
new_name = os.readlink(boot_file)
if new_name[0] == "/":
boot_file = new_name
else:
path = boot_file.split('/')
path[len(path)-1] = new_name
boot_file = '/'.join(path)
if not os.path.exists(boot_file):
err("Boot file \'" + boot_file + "\' not found.")
grub_fd = open(boot_file)
for line in grub_fd:
if xen_title_re.match(line):
within_xen_title = 1
elif within_xen_title and xen_kernel_re.match(line):
within_xen_entry = 1
elif within_xen_title and within_xen_entry and kernel_ver_re.match(line):
for i in line.split():
if (i.find("vmlinuz-") >= 0):
if kernel_version == i[i.find("vmlinuz-") + len("vmlinuz-"):]:
insert_at_end_of_entry = 1
path_prefix = i[0:i.find("vmlinuz-")]
elif any_module_re.match(line) and insert_at_end_of_entry:
if binary_name_re.match(line):
#delete existing policy module line
line=''
elif any_title_re.match(line):
within_xen_title = 0
within_xen_entry = 0
if (empty_line_re.match(line) or any_title_re.match(line)) and insert_at_end_of_entry:
#newline or new title: we insert the policy module line here
os.write(tmp_fd, "\tmodule " + path_prefix + policy_name + ".bin\n")
insert_at_end_of_entry = 0
#write the line that was read (except potential existing policy entry)
os.write(tmp_fd, line)
if insert_at_end_of_entry:
#last entry, no empty line at end of file
os.write(tmp_fd, "\tmodule " + path_prefix + policy_name + ".bin\n")
#temp file might be destroyed when closing it, first copy ...
shutil.move(boot_file, boot_file+"_save")
shutil.copyfile(tmp_grub, boot_file)
os.close(tmp_fd)
#temp file did not disappear on my system ...
try:
os.remove(tmp_grub)
except:
pass
def main(argv):
try:
user_kver = None
policy = None
if len(argv) == 2:
policy = argv[1]
elif len(argv) == 3:
policy = argv[1]
user_kver = argv[2]
else:
usage()
if not policy_name_re.match(policy):
err("Illegal policy name \'" + policy + "\'")
policy_file = policy_dir_prefix + "/" + string.join(string.split(policy, "."), "/")
src_binary_policy_file = policy_file + ".bin"
#check if .bin exists or if policy file exists
if not os.path.isfile(src_binary_policy_file):
if not os.path.isfile(policy_file + "-security_policy.xml"):
err("Unknown policy \'" + policy +"\'")
else:
err("Cannot find binary file for policy \'" + policy +
"\'. Please use makepolicy to create binary file.")
dst_binary_policy_file = "/boot/" + policy + ".bin"
shutil.copyfile(src_binary_policy_file, dst_binary_policy_file)
kernel_version = determine_kernelversion(user_kver)
insert_policy(boot_filename, kernel_version, policy)
print "Boot entry created and \'%s\' copied to /boot" % (policy + ".bin")
except ACMError:
pass
except:
traceback.print_exc(limit=1)
if __name__ == '__main__':
main(sys.argv)
tools/python/xen/xm/dumppolicy.py
+49
View File
@@ -0,0 +1,49 @@
#============================================================================
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#============================================================================
# Copyright (C) 2006 International Business Machines Corp.
# Author: Reiner Sailer <sailer@us.ibm.com>
#============================================================================
"""Display currently enforced policy (low-level hypervisor representation).
"""
import sys
import traceback
import os
import commands
import shutil
import string
from xen.util.security import ACMError, err, dump_policy
def usage():
print "\nUsage: xm dumppolicy\n"
print " Retrieve and print currently enforced"
print " hypervisor policy information (low-level).\n"
err("Usage")
def main(argv):
try:
dump_policy()
except ACMError:
pass
except:
traceback.print_exc(limit=1)
if __name__ == '__main__':
main(sys.argv)
tools/python/xen/xm/labels.py
+85
View File
@@ -0,0 +1,85 @@
#============================================================================
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#============================================================================
# Copyright (C) 2006 International Business Machines Corp.
# Author: Reiner Sailer <sailer@us.ibm.com>
#============================================================================
"""Listing available labels for a policy.
"""
import sys
import traceback
import os
import commands
import shutil
import string
from xen.util.security import ACMError, err, list_labels, active_policy
from xen.util.security import vm_label_re, res_label_re, all_label_re
def usage():
print "\nUsage: xm labels [<policy>] [<type=dom|res|any>]\n"
print " Prints labels of the specified type (default is dom)"
print " that are defined in policy (default is current"
print " hypervisor policy).\n"
err("Usage")
def main(argv):
try:
policy = None
type = None
for i in argv[1:]:
i_s = string.split(i, '=')
if len(i_s) > 1:
if (i_s[0] == 'type') and (len(i_s) == 2):
if not type:
type = i_s[1]
else:
usage()
else:
usage()
else:
if not policy:
policy = i
else:
usage()
if not policy:
policy = active_policy
if active_policy in ['NULL', 'INACTIVE', 'DEFAULT']:
err("No policy active. Please specify the <policy> parameter.")
if not type or (type in ['DOM', 'dom']):
condition = vm_label_re
elif type in ['RES', 'res']:
condition = res_label_re
elif type in ['ANY', 'any']:
condition = all_label_re
else:
err("Unknown label type \'" + type + "\'")
labels = list_labels(policy, condition)
labels.sort()
for label in labels:
print label
except ACMError:
pass
except:
traceback.print_exc(limit=1)
if __name__ == '__main__':
main(sys.argv)
tools/python/xen/xm/loadpolicy.py
+51
View File
@@ -0,0 +1,51 @@
#============================================================================
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#============================================================================
# Copyright (C) 2006 International Business Machines Corp.
# Author: Reiner Sailer <sailer@us.ibm.com>
#============================================================================
"""Loading a compiled binary policy into the hypervisor.
"""
import sys
import traceback
import os
import commands
import shutil
import string
from xen.util.security import ACMError, err, load_policy
def usage():
print "\nUsage: xm loadpolicy <policy>\n"
print " Load the compiled binary (.bin) policy"
print " into the running hypervisor.\n"
err("Usage")
def main(argv):
try:
if len(argv) != 2:
usage()
load_policy(argv[1])
except ACMError:
pass
except:
traceback.print_exc(limit=1)
if __name__ == '__main__':
main(sys.argv)
tools/python/xen/xm/makepolicy.py
+53
View File
@@ -0,0 +1,53 @@
#============================================================================
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#============================================================================
# Copyright (C) 2006 International Business Machines Corp.
# Author: Reiner Sailer <sailer@us.ibm.com>
#============================================================================
"""Compiling a XML source policy file into mapping and binary versions.
"""
import sys
import traceback
import os
import commands
import shutil
import string
from xen.util.security import ACMError, err, make_policy
def usage():
print "\nUsage: xm makepolicy <policy>\n"
print " Translate an XML source policy and create"
print " mapping file and binary policy.\n"
err("Usage")
def main(argv):
try:
if len(argv) != 2:
usage()
make_policy(argv[1])
except ACMError:
pass
except:
traceback.print_exc(limit=1)
if __name__ == '__main__':
main(sys.argv)
Reference in new issue
xen-project/xen
Block a user
Blocking a user prevents them from interacting with repositories, such as opening or commenting on pull requests or issues. Learn more about blocking a user.

The note is not visible to the blocked user.