QubesOS/qubes-core-admin-linux
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

qrexec: reformat qrexec-policy

Browse Source 
No functional change, just make it slightly less painful to read...
This commit is contained in:
Marek Marczykowski-Górecki 2016-08-16 02:57:17 +02:00
parent 92c3ba578a
commit 8a780cb7f5
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
1 changed files with 100 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
qrexec/qrexec-policy
View File

@ -9,18 +9,19 @@ from optparse import OptionParser
import fcntl import fcntl
from PyQt4.QtGui import QApplication, QMessageBox from PyQt4.QtGui import QApplication, QMessageBox
POLICY_FILE_DIR = "/etc/qubes-rpc/policy" POLICY_FILE_DIR = "/etc/qubes-rpc/policy"
# XXX: Backward compatibility, to be removed soon # XXX: Backward compatibility, to be removed soon
DEPRECATED_POLICY_FILE_DIR = "/etc/qubes_rpc/policy" DEPRECATED_POLICY_FILE_DIR = "/etc/qubes_rpc/policy"
QREXEC_CLIENT = "/usr/lib/qubes/qrexec-client" QREXEC_CLIENT = "/usr/lib/qubes/qrexec-client"
QUBES_RPC_MULTIPLEXER_PATH = "/usr/lib/qubes/qubes-rpc-multiplexer" QUBES_RPC_MULTIPLEXER_PATH = "/usr/lib/qubes/qubes-rpc-multiplexer"
class UserChoice: class UserChoice:
ALLOW = 0 ALLOW = 0
DENY = 1 DENY = 1
ALWAYS_ALLOW = 2 ALWAYS_ALLOW = 2
def prepare_app(): def prepare_app():
app = QApplication(sys.argv) app = QApplication(sys.argv)
app.setOrganizationName("The Qubes Project") app.setOrganizationName("The Qubes Project")
@ -28,6 +29,7 @@ def prepare_app():
app.setApplicationName("Qubes") app.setApplicationName("Qubes")
return app return app
def ask(text, title="Question", yestoall=False): def ask(text, title="Question", yestoall=False):
prepare_app() prepare_app()
@ -35,7 +37,8 @@ def ask(text, title="Question", yestoall=False):
if yestoall: if yestoall:
buttons |= QMessageBox.YesToAll buttons |= QMessageBox.YesToAll
reply = QMessageBox.question(None, title, text, buttons, defaultButton=QMessageBox.Yes) reply = QMessageBox.question(None, title, text, buttons,
defaultButton=QMessageBox.Yes)
if reply == QMessageBox.Yes: if reply == QMessageBox.Yes:
return 0 return 0
elif reply == QMessageBox.No: elif reply == QMessageBox.No:
@ -46,6 +49,7 @@ def ask(text, title="Question", yestoall=False):
# ?! # ?!
return 127 return 127
def line_to_dict(line): def line_to_dict(line):
tokens = line.split() tokens = line.split()
if len(tokens) < 3: if len(tokens) < 3:
@ -54,24 +58,25 @@ def line_to_dict(line):
if tokens[0][0] == '#': if tokens[0][0] == '#':
return None return None
dict={} policy_dict = {
dict['source']=tokens[0] 'source': tokens[0],
dict['dest']=tokens[1] 'dest': tokens[1],
'full-action': tokens[2],
}
dict['full-action']=tokens[2]
action_list = tokens[2].split(',') action_list = tokens[2].split(',')
dict['action']=action_list.pop(0) policy_dict['action'] = action_list.pop(0)
for iter in action_list: for action_iter in action_list:
paramval=iter.split("=") paramval = action_iter.split("=")
dict["action."+paramval[0]]=paramval[1] policy_dict["action." + paramval[0]] = paramval[1]
# Warn if we're ignoring extra data after a space, such as: # Warn if we're ignoring extra data after a space, such as:
# vm1 vm2 allow, user=foo # vm1 vm2 allow, user=foo
if len(tokens) > 3: if len(tokens) > 3:
print >> sys.stderr, "Trailing data ignored in %s" % line print >> sys.stderr, "Trailing data ignored in %s" % line
return dict return policy_dict
def read_policy_file(service_name): def read_policy_file(service_name):
@ -83,35 +88,39 @@ def read_policy_file(service_name):
policy_file = os.path.join(DEPRECATED_POLICY_FILE_DIR, service_name) policy_file = os.path.join(DEPRECATED_POLICY_FILE_DIR, service_name)
if not os.path.isfile(policy_file): if not os.path.isfile(policy_file):
return None return None
print >>sys.stderr, "RPC service '%s' uses deprecated policy location, please move to %s" % (service_name, POLICY_FILE_DIR) print >> sys.stderr, \
"RPC service '%s' uses deprecated policy location, " \
"please move to %s" % (service_name, POLICY_FILE_DIR)
policy_list = list() policy_list = list()
f = open(policy_file) f = open(policy_file)
fcntl.flock(f, fcntl.LOCK_SH) fcntl.flock(f, fcntl.LOCK_SH)
for iter in f.readlines(): for policy_iter in f.readlines():
dict = line_to_dict(iter) policy_item = line_to_dict(policy_iter)
if dict is not None: if policy_item is not None:
policy_list.append(dict) policy_list.append(policy_item)
f.close() f.close()
return policy_list return policy_list
def is_match(item, config_term): def is_match(item, config_term):
return (item is not "dom0" and config_term == "$anyvm") or item == config_term return (item is not "dom0" and config_term == "$anyvm") or \
item == config_term
def get_default_policy(): def get_default_policy():
dict={} return {"action": "deny"}
dict["action"]="deny"
return dict
def find_policy(policy, domain, target): def find_policy(policy, domain, target):
for iter in policy: for policy_iter in policy:
if not is_match(domain, iter["source"]): if not is_match(domain, policy_iter["source"]):
continue continue
if not is_match(target, iter["dest"]): if not is_match(target, policy_iter["dest"]):
continue continue
return iter return policy_iter
return get_default_policy() return get_default_policy()
def validate_target(target): def validate_target(target):
# special targets # special targets
if target in ['$dispvm']: if target in ['$dispvm']:
@ -121,6 +130,7 @@ def validate_target(target):
return app.domains[target] return app.domains[target]
def spawn_target_if_necessary(vm): def spawn_target_if_necessary(vm):
if vm.is_running(): if vm.is_running():
return return
@ -131,9 +141,11 @@ def spawn_target_if_necessary(vm):
stdin=null, stdout=null) stdin=null, stdout=null)
null.close() null.close()
def do_execute(domain, target, user, service_name, process_ident, vm=None): def do_execute(domain, target, user, service_name, process_ident, vm=None):
if target == "$dispvm": if target == "$dispvm":
cmd = "/usr/lib/qubes/qfile-daemon-dvm " + service_name + " " + domain + " " +user cmd = "/usr/lib/qubes/qfile-daemon-dvm " + service_name + " " + \
domain + " " + user
os.execl(QREXEC_CLIENT, "qrexec-client", os.execl(QREXEC_CLIENT, "qrexec-client",
"-d", "dom0", "-c", process_ident, cmd) "-d", "dom0", "-c", process_ident, cmd)
else: else:
@ -149,12 +161,15 @@ def do_execute(domain, target, user, service_name, process_ident, vm=None):
os.execl(QREXEC_CLIENT, "qrexec-client", os.execl(QREXEC_CLIENT, "qrexec-client",
"-d", target, "-c", process_ident, cmd) "-d", target, "-c", process_ident, cmd)
def confirm_execution(domain, target, service_name): def confirm_execution(domain, target, service_name):
text = "Do you allow domain \"" + domain + "\" to execute " + service_name text = "Do you allow domain \"" + domain + "\" to execute " + service_name
text += " operation on the domain \"" + target + "\"?<br>" text += " operation on the domain \"" + target + "\"?<br>"
text+= " \"Yes to All\" option will automatically allow this operation in the future." text += " \"Yes to All\" option will automatically allow this " \
"operation in the future."
return qubes.guihelpers.ask(text, yestoall=True) return qubes.guihelpers.ask(text, yestoall=True)
def add_always_allow(domain, target, service_name, options): def add_always_allow(domain, target, service_name, options):
policy_file = POLICY_FILE_DIR + "/" + service_name policy_file = POLICY_FILE_DIR + "/" + service_name
if not os.path.isfile(policy_file): if not os.path.isfile(policy_file):
@ -169,6 +184,7 @@ def add_always_allow(domain, target, service_name, options):
f.write("".join(lines)) f.write("".join(lines))
f.close() f.close()
def info_dialog(msg_type, text): def info_dialog(msg_type, text):
if msg_type not in ['info', 'warning', 'error']: if msg_type not in ['info', 'warning', 'error']:
raise ValueError("Invalid msg_type value") raise ValueError("Invalid msg_type value")
@ -184,18 +200,24 @@ def info_dialog(msg_type, text):
subprocess.call(["/usr/bin/kdialog", "--{}".format(kdialog_msg_type), subprocess.call(["/usr/bin/kdialog", "--{}".format(kdialog_msg_type),
text]) text])
# noinspection PyUnusedLocal
def policy_editor(domain, target, service_name): def policy_editor(domain, target, service_name):
text = "No policy definition found for " + service_name + " action. " text = "No policy definition found for " + service_name + " action. "
text += "Please create a policy file in Dom0 in " + POLICY_FILE_DIR + "/" + service_name text += "Please create a policy file in Dom0 in " + POLICY_FILE_DIR + "/" + service_name
info_dialog("warning", text) info_dialog("warning", text)
def main(): def main():
usage = "usage: %prog [options] <src-domain-id> <src-domain> <target-domain> <service> <process-ident>" usage = "usage: %prog [options] <src-domain-id> <src-domain> <target-domain> <service> <process-ident>"
parser = OptionParser(usage) parser = OptionParser(usage)
parser.add_option ("--assume-yes-for-ask", action="store_true", dest="assume_yes_for_ask", default=False, parser.add_option("--assume-yes-for-ask", action="store_true",
dest="assume_yes_for_ask", default=False,
help="Allow run of service without confirmation if policy say 'ask'") help="Allow run of service without confirmation if policy say 'ask'")
parser.add_option ("--just-evaluate", action="store_true", dest="just_evaluate", default=False, parser.add_option("--just-evaluate", action="store_true",
help="Do not run the service, only evaluate policy; retcode=0 means 'allow'") dest="just_evaluate", default=False,
help="Do not run the service, only evaluate policy; "
"retcode=0 means 'allow'")
(options, args) = parser.parse_args() (options, args) = parser.parse_args()
domain_id = args[0] domain_id = args[0]
@ -211,17 +233,18 @@ def main():
try: try:
vm = validate_target(target) vm = validate_target(target)
except KeyError: except KeyError:
print >> sys.stderr, "Rpc failed (unknown domain):", domain, target, service_name print >> sys.stderr, "Rpc failed (unknown domain):", \
domain, target, service_name
text = "Domain '%s' doesn't exist (service %s called by domain %s)." % ( text = "Domain '%s' doesn't exist (service %s called by domain %s)." % (
target, service_name, domain) target, service_name, domain)
info_dialog("error", text) info_dialog("error", text)
exit(1) exit(1)
policy_list = read_policy_file(service_name) policy_list = read_policy_file(service_name)
if policy_list==None: if policy_list is None:
policy_editor(domain, target, service_name) policy_editor(domain, target, service_name)
policy_list = read_policy_file(service_name) policy_list = read_policy_file(service_name)
if policy_list==None: if policy_list is None:
policy_list = list() policy_list = list()
policy_dict = find_policy(policy_list, domain, target) policy_dict = find_policy(policy_list, domain, target)
@ -232,7 +255,8 @@ def main():
if policy_dict["action"] == "ask": if policy_dict["action"] == "ask":
user_choice = confirm_execution(domain, target, service_name) user_choice = confirm_execution(domain, target, service_name)
if user_choice == UserChoice.ALWAYS_ALLOW: if user_choice == UserChoice.ALWAYS_ALLOW:
add_always_allow(domain, target, service_name, policy_dict["full-action"].lstrip('ask')) add_always_allow(domain, target, service_name,
policy_dict["full-action"].lstrip('ask'))
policy_dict["action"] = "allow" policy_dict["action"] = "allow"
elif user_choice == UserChoice.ALLOW: elif user_choice == UserChoice.ALLOW:
policy_dict["action"] = "allow" policy_dict["action"] = "allow"
@ -246,9 +270,9 @@ def main():
exit(1) exit(1)
if policy_dict["action"] == "allow": if policy_dict["action"] == "allow":
if policy_dict.has_key("action.target"): if "action.target" in policy_dict:
target = policy_dict["action.target"] target = policy_dict["action.target"]
if policy_dict.has_key("action.user"): if "action.user" in policy_dict:
user = policy_dict["action.user"] user = policy_dict["action.user"]
else: else:
user = "DEFAULT" user = "DEFAULT"
@ -258,4 +282,5 @@ def main():
print >> sys.stderr, "Rpc denied:", domain, target, service_name print >> sys.stderr, "Rpc denied:", domain, target, service_name
exit(1) exit(1)
main() main()