CORD-1392: Autogenerate security policies
Change-Id: Ia183f2a84b27923802e62435c82b20b50fb3fcee
diff --git a/lib/xos-genx/tests/general_security_test.py b/lib/xos-genx/tests/general_security_test.py
new file mode 100644
index 0000000..f2dd857
--- /dev/null
+++ b/lib/xos-genx/tests/general_security_test.py
@@ -0,0 +1,165 @@
+import unittest
+from xosgenx.generator import XOSGenerator
+from helpers import FakeArgs, XProtoTestHelpers
+import pdb
+
+"""The function below is for eliminating warnings arising due to the missing policy_output_0,
+which is generated and loaded dynamically.
+"""
+def policy_output_0(x, y):
+ raise Exception("Security enforcer not generated. Test failed.")
+ return False
+
+"""
+The tests below use the Python code target to generate
+Python security policies, set up an appropriate environment and execute the Python.
+"""
+class XProtoSecurityTest(unittest.TestCase):
+ def setUp(self):
+ self.target = XProtoTestHelpers.write_tmp_target("{{ xproto_fol_to_python_test(proto.policies.test_policy, None, '0') }}")
+
+ def test_constant(self):
+ xproto = \
+"""
+ policy test_policy < True >
+"""
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = self.target
+
+ output = XOSGenerator.generate(args)
+
+ exec(output) # This loads the generated function, which should look like this:
+
+ """
+ def policy_output_0(obj, ctx):
+ i1 = True
+ return i1
+ """
+
+ verdict = policy_output_0({}, {})
+ self.assertTrue(verdict)
+
+ def test_equal(self):
+ xproto = \
+"""
+ policy test_policy < ctx.user = obj.user >
+"""
+
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = self.target
+
+ output = XOSGenerator.generate(args)
+
+ exec(output) # This loads the generated function, which should look like this:
+
+ """
+ def policy_output_0(obj, ctx):
+ i1 = (ctx.user == obj.user)
+ return i1
+ """
+
+ obj = FakeArgs()
+ obj.user = 1
+ ctx = FakeArgs()
+ ctx.user = 1
+
+ verdict = policy_output_0(obj, ctx)
+
+ def test_bin(self):
+ xproto = \
+"""
+ policy test_policy < ctx.is_admin = True | obj.empty = True>
+"""
+
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = self.target
+
+ output = XOSGenerator.generate(args)
+ exec(output) # This loads the generated function, which should look like this:
+
+ """
+ def policy_output_0(obj, ctx):
+ i2 = (ctx.is_admin == True)
+ i3 = (obj.empty == True)
+ i1 = (i2 or i3)
+ return i1
+ """
+
+ obj = FakeArgs()
+ obj.empty = True
+
+ ctx = FakeArgs()
+ ctx.is_admin = True
+
+ verdict = policy_output_0(obj, ctx)
+
+ self.assertTrue(verdict)
+
+
+ def test_exists(self):
+ xproto = \
+"""
+ policy test_policy < exists Privilege: Privilege.object_id = obj.id >
+"""
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = self.target
+
+ output = XOSGenerator.generate(args)
+ exec(output) # This loads the generated function, which should look like this:
+
+ """
+ def policy_output_0(obj, ctx):
+ i1 = Privilege.objects.filter(object_id=obj.id)
+ return i1
+ """
+
+ self.assertTrue(policy_output_0 is not None)
+
+ def test_python(self):
+ xproto = \
+"""
+ policy test_policy < {{ "jack" in ["the", "box"] }} = False >
+"""
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = self.target
+ output = XOSGenerator.generate(args)
+ exec(output) # This loads the generated function, which should look like this:
+
+ """
+ def policy_output_0(obj, ctx):
+ i2 = ('jack' in ['the', 'box'])
+ i1 = (i2 == False)
+ return i1
+ """
+
+ self.assertTrue(policy_output_0({}, {}) is True)
+
+ def test_forall(self):
+ # This one we only parse
+ xproto = \
+"""
+ policy test_policy < forall Credential: Credential.obj_id = obj_id >
+"""
+
+ target = XProtoTestHelpers.write_tmp_target("{{ xproto_fol_to_python_test(proto.policies.test_policy, None, '0') }}")
+
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = target
+
+ output = XOSGenerator.generate(args)
+ """
+ def policy_output_0(obj, ctx):
+ i2 = Credential.objects.filter((~ Q(obj_id=obj_id)))[0]
+ i1 = (not i2)
+ return i1
+ """
+ exec(output)
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/lib/xos-genx/tests/optimize_test.py b/lib/xos-genx/tests/optimize_test.py
new file mode 100644
index 0000000..ce15bf4
--- /dev/null
+++ b/lib/xos-genx/tests/optimize_test.py
@@ -0,0 +1,74 @@
+import unittest
+from xosgenx.jinja2_extensions.fol2 import FOL2Python
+import pdb
+
+class XProtoOptimizeTest(unittest.TestCase):
+ def setUp(self):
+ self.f2p = FOL2Python()
+
+ def test_constant(self):
+ input = 'True'
+ output = self.f2p.hoist_constants(input)
+ self.assertEqual(output, input)
+
+ def test_exists(self):
+ input = {'exists': ['x',{'|':['x','y']}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'|': ['y', {'exists': ['x', 'x']}]}
+ self.assertEqual(output, expected)
+
+ def test_forall(self):
+ input = {'forall': ['x',{'|':['x','y']}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'|': ['y', {'forall': ['x', 'x']}]}
+ self.assertEqual(output, expected)
+
+ def test_exists_embedded(self):
+ input = {'|':['True',{'exists': ['x',{'|':['x','y']}]}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'|': ['True', {'|': ['y', {'exists': ['x', 'x']}]}]}
+ self.assertEqual(output, expected)
+
+ def test_exists_equals(self):
+ input = {'|':['True',{'exists': ['x',{'|':['x',{'=':['y','z']}]}]}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'|': ['True', {'|': [{'=': ['y', 'z']}, {'exists': ['x', 'x']}]}]}
+ self.assertEqual(output, expected)
+
+ def test_exists_nested_constant(self):
+ input = {'|':['True',{'exists': ['x',{'|':['y',{'=':['y','x']}]}]}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'|': ['True', {'|': ['y', {'exists': ['x', {'=': ['y', 'x']}]}]}]}
+ self.assertEqual(output, expected)
+
+ def test_exists_nested(self):
+ input = {'exists': ['x',{'exists':['y',{'=':['y','x']}]}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'exists': ['x', {'exists': ['y', {'=': ['y', 'x']}]}]}
+ self.assertEqual(output, expected)
+
+ def test_exists_nested2(self):
+ input = {'exists': ['x',{'exists':['y',{'=':['z','y']}]}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'exists': ['y', {'=': ['z', 'y']}]}
+ self.assertEqual(output, expected)
+
+ def test_exists_nested3(self):
+ input = {'exists': ['x',{'exists':['y',{'=':['z','x']}]}]}
+
+ output = self.f2p.hoist_constants(input)
+ expected = {'exists': ['x', {'=': ['z', 'x']}]}
+ self.assertEqual(output, expected)
+
+
+if __name__ == '__main__':
+ unittest.main()
+
+
diff --git a/lib/xos-genx/tests/policy_test.py b/lib/xos-genx/tests/policy_test.py
index 5f1dca0..fcfed8a 100644
--- a/lib/xos-genx/tests/policy_test.py
+++ b/lib/xos-genx/tests/policy_test.py
@@ -1,7 +1,6 @@
import unittest
from xosgenx.generator import XOSGenerator
from helpers import FakeArgs, XProtoTestHelpers
-import pdb
"""
The tests below convert the policy logic expression
@@ -9,6 +8,25 @@
"""
class XProtoPolicyTest(unittest.TestCase):
+ def test_annotation(self):
+ xproto = \
+"""
+ policy true_policy < True >
+
+ message always::true_policy {
+ required int still = 9;
+ }
+"""
+
+ target = XProtoTestHelpers.write_tmp_target("{{ proto.messages.0 }}")
+
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = target
+
+ output = XOSGenerator.generate(args)
+ self.assertIn("true_policy", output)
+
def test_constant(self):
xproto = \
"""
@@ -24,6 +42,66 @@
output = XOSGenerator.generate(args).replace('t','T')
self.assertTrue(eval(output))
+ def test_term(self):
+ xproto = \
+"""
+ policy slice_user < slice.user.is_admin >
+"""
+
+ target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = target
+
+ output = XOSGenerator.generate(args)
+
+ slice = FakeArgs()
+ slice.user = FakeArgs()
+ slice.user.is_admin = True
+
+ expr = eval(output)
+ self.assertTrue(expr)
+
+ def test_num_constant(self):
+ xproto = \
+"""
+ policy slice_user < slice.user.age = 57 >
+"""
+
+ target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = target
+
+ output = XOSGenerator.generate(args)
+
+ slice = FakeArgs()
+ slice.user = FakeArgs()
+ slice.user.is_admin = True
+
+ expr = eval(output)
+ self.assertTrue(expr)
+
+ def test_string_constant(self):
+ xproto = \
+"""
+ policy slice_user < slice.user.email = "admin@opencord.org" >
+"""
+
+ target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = target
+
+ output = XOSGenerator.generate(args)
+
+ slice = FakeArgs()
+ slice.user = FakeArgs()
+ slice.user.is_admin = True
+
+ expr = eval(output)
+ self.assertTrue(expr)
+
def test_equal(self):
xproto = \
"""
diff --git a/lib/xos-genx/tests/xos_security_test.py b/lib/xos-genx/tests/xos_security_test.py
new file mode 100644
index 0000000..30a7193
--- /dev/null
+++ b/lib/xos-genx/tests/xos_security_test.py
@@ -0,0 +1,77 @@
+import unittest
+from xosgenx.generator import XOSGenerator
+from helpers import FakeArgs, XProtoTestHelpers
+import pdb
+
+"""The function below is for eliminating warnings arising due to the missing policy_output_0,
+which is generated and loaded dynamically.
+"""
+def policy_output_0(x, y):
+ raise Exception("Security enforcer not generated. Test failed.")
+ return False
+
+"""
+The tests below use the Python code target to generate
+Python security policies, set up an appropriate environment and execute the Python.
+"""
+class XProtoXOSSecurityTest(unittest.TestCase):
+ def setUp(self):
+ self.target = XProtoTestHelpers.write_tmp_target("{{ xproto_fol_to_python_test(proto.policies.test_policy, None, '0') }}")
+
+ """
+ This is the security policy for controllers
+ """
+ def test_controller_policy(self):
+ xproto = \
+"""
+ policy test_policy < ctx.user.is_admin | exists Privilege: Privilege.user_id = ctx.user.id & Privilege.object_type = "Deployment" >
+"""
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = self.target
+
+ output = XOSGenerator.generate(args)
+
+ exec(output) # This loads the generated function, which should look like this:
+
+ """
+ def policy_output_0(obj, ctx):
+ i2 = ctx.user.is_admin
+ i3 = Privilege.objects.filter(Q(user_id=ctx.user.id), Q(object_type='Deployment'))[0]
+ i1 = (i2 or i3)
+ return i1
+ """
+
+ # FIXME: Test this policy by executing it
+ self.assertTrue(policy_output_0 is not None)
+
+ """
+ This is the security policy for controllers
+ """
+ def _test_controller_network_policy(self):
+ xproto = \
+"""
+ policy test_policy < ctx.user.is_admin | exists Slice: forall ctx.networks: ctx.networks.owner.id = Slice.id >
+"""
+ args = FakeArgs()
+ args.inputs = xproto
+ args.target = self.target
+
+ output = XOSGenerator.generate(args)
+
+ exec(output) # This loads the generated function, which should look like this:
+
+ """
+ def policy_output_0(obj, ctx):
+ i2 = ctx.user.is_admin
+ i3 = Privilege.objects.filter(Q(user_id=ctx.user.id), Q(object_type='Deployment'))[0]
+ i1 = (i2 or i3)
+ return i1
+ """
+
+ # FIXME: Test this policy by executing it
+ self.assertTrue(policy_output_0 is not None)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/lib/xos-genx/xosgenx/jinja2_extensions/__init__.py b/lib/xos-genx/xosgenx/jinja2_extensions/__init__.py
index 659f440..ea30100 100644
--- a/lib/xos-genx/xosgenx/jinja2_extensions/__init__.py
+++ b/lib/xos-genx/xosgenx/jinja2_extensions/__init__.py
@@ -1,2 +1,3 @@
from .django import *
from .base import *
+from .fol2 import *
diff --git a/lib/xos-genx/xosgenx/jinja2_extensions/fol2.py b/lib/xos-genx/xosgenx/jinja2_extensions/fol2.py
new file mode 100644
index 0000000..a929a9c
--- /dev/null
+++ b/lib/xos-genx/xosgenx/jinja2_extensions/fol2.py
@@ -0,0 +1,382 @@
+import astunparse
+import ast
+import random
+import string
+from plyxproto.parser import *
+import pdb
+
+BINOPS = ['|', '&', '=>']
+QUANTS = ['exists', 'forall']
+
+
+class PolicyException(Exception):
+ pass
+
+
+class AutoVariable:
+ def __init__(self, base):
+ self.base = base
+
+ def __iter__(self):
+ self.idx = 0
+ return self
+
+ def next(self):
+ var = 'i%d' % self.idx
+ self.idx += 1
+ return var
+
+
+def gen_random_string():
+ return ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(5))
+
+
+class FOL2Python:
+ def __init__(self, context_map=None):
+ # This will produce i0, i1, i2 etc.
+ self.loopvar = iter(AutoVariable('i'))
+ self.verdictvar = iter(AutoVariable('result'))
+
+ self.loop_variable = self.loopvar.next()
+ self.verdict_variable = self.verdictvar.next()
+ self.context_map = context_map
+
+ if not self.context_map:
+ self.context_map = {'user': 'self', 'obj': 'obj'}
+
+ def loop_next(self):
+ self.loop_variable = self.loopvar.next()
+
+ def verdict_next(self):
+ self.verdict_variable = self.verdictvar.next()
+
+ def gen_enumerate(self, fol):
+ pass
+
+ def format_term_for_query(self, model, term, django=False):
+ if term.startswith(model + '.'):
+ term = term[len(model) + 1:]
+ if django:
+ term = term.replace('.', '__')
+ else:
+ term = '__elt' + '.' + term
+ return term
+
+ def fol_to_python_filter(self, model, e, django=False, negate=False):
+ try:
+ (k, v), = e.items()
+ except AttributeError:
+ return [self.format_term_for_query(model, e)]
+
+ if django:
+ if negate:
+ # De Morgan's negation
+ q_bracket = '~Q(%s)'
+ or_expr = ','
+ and_expr = '|'
+ else:
+ q_bracket = 'Q(%s)'
+ or_expr = '|'
+ and_expr = ','
+ else:
+ if negate:
+ # De Morgan's negation
+ q_bracket = 'not %s'
+ or_expr = ' and '
+ and_expr = ' or '
+ else:
+ q_bracket = '%s'
+ or_expr = ' or '
+ and_expr = ' and '
+
+ if k == '=':
+ v = [self.format_term_for_query(
+ model, term, django=django) for term in v]
+ if django:
+ operator = ' = '
+ else:
+ operator = ' == '
+ return [q_bracket % operator.join(v)]
+ elif k == '|':
+ components = [self.fol_to_python_filter(
+ model, x, django=django).pop() for x in v]
+ return [or_expr.join(components)]
+ elif k == '&':
+ components = [self.fol_to_python_filter(
+ model, x, django=django).pop() for x in v]
+ return [and_expr.join(components)]
+ elif k == '=>':
+ components = [self.fol_to_python_filter(
+ model, x, django=django).pop() for x in v]
+ return ['~%s | %s' % (components[0], components[1])]
+
+ """ Convert a single leaf node from a string
+ to an AST"""
+
+ def str_to_ast(self, s):
+ ast_module = ast.parse(s)
+ return ast_module.body[0]
+
+ def hoist_constants(self, fol, var=None):
+ try:
+ (k, v), = fol.items()
+ except AttributeError:
+ k = 'term'
+ v = fol
+
+ if k == 'python':
+ # Tainted, don't optimize
+ if var:
+ return {'hoist': []}
+ else:
+ return fol
+
+ if var:
+ if k == 'term':
+ if not v.startswith(var):
+ return {'hoist': ['const', fol], 'result': 'True'}
+ else:
+ return {'hoist': [], 'result': fol}
+ elif k == '=':
+ lhs, rhs = v
+ if not lhs.startswith(var) and not rhs.startswith(var):
+ return {'hoist': ['=', fol], 'result': 'True'} # XXX
+ else:
+ return {'hoist': [], 'result': fol}
+ elif k in BINOPS:
+ lhs, rhs = v
+ rlhs = self.hoist_constants(lhs, var)
+ rrhs = self.hoist_constants(rhs, var)
+
+ if rlhs['hoist'] and rrhs['hoist']:
+ return {'hoist': ['=', fol], 'result': 'True'}
+ elif rlhs['hoist']:
+ return {'hoist': [k, lhs], 'result': rhs}
+ elif rrhs['hoist']:
+ return {'hoist': [k, rhs], 'result': lhs}
+ else:
+ return {'hoist': [], 'result': fol}
+
+ elif k in QUANTS:
+ var2, expr = v
+ result = self.hoist_constants(expr, var2)
+ if result['hoist']:
+ if result['result'] == 'True':
+ return {'hoist': ['const'], 'result': result['hoist'][1]}
+ elif result['hoist'][0] in BINOPS:
+ return {'hoist': ['const'], 'result': {result['hoist'][0]: [result['hoist'][1], {k: [var2, result['result']]}]}}
+ else:
+ return {'hoist': ['const'], 'result': {k: [var2, result['result']]}}
+ else:
+ result = self.hoist_constants(expr, var)
+ if result['result'] == 'True':
+ return {'hoist': ['&', fol], 'result': 'True'}
+ else:
+ return {'hoist': [], 'result': fol}
+ else:
+ return {'hoist': [], 'result': fol}
+ else:
+ if k in BINOPS:
+ lhs, rhs = v
+ rlhs = self.hoist_constants(lhs)
+ rrhs = self.hoist_constants(rhs)
+ return {k: [rlhs, rrhs]}
+ elif k in QUANTS:
+ var, expr = v
+ result = self.hoist_constants(expr, var)
+ if result['hoist']:
+ if result['result'] == 'True':
+ return result['hoist'][1]
+ elif result['hoist'][0] in BINOPS:
+ return {result['hoist'][0]: [result['hoist'][1], {k: [var, result['result']]}]}
+ else:
+ return {k: [var, result['result']]}
+ else:
+ return fol
+ else:
+ return fol
+
+ def gen_test_function(self, fol, policy_name, tag):
+ if not tag:
+ tag = gen_random_string()
+
+ policy_function_name = 'policy_%(policy_name)s_%(random_string)s' % {
+ 'policy_name': policy_name, 'random_string': tag}
+ self.verdict_next()
+ function_str = """
+def %(fn_name)s(obj, ctx):
+ return %(vvar)s
+ """ % {'fn_name': policy_function_name, 'vvar': self.verdict_variable}
+
+ function_ast = self.str_to_ast(function_str)
+ policy_code = self.gen_test(fol, self.verdict_variable)
+
+ function_ast.body = [policy_code] + function_ast.body
+
+ return astunparse.unparse(function_ast)
+
+ def gen_test(self, fol, verdict_var, bindings=None):
+ if isinstance(fol, str):
+ return self.str_to_ast('%(verdict_var)s = %(constant)s' % {'verdict_var': verdict_var, 'constant': fol})
+
+ (k, v), = fol.items()
+
+ if k == '=':
+ # This is the simplest case, we don't recurse further
+ # To use terms that are not simple variables, use
+ # the Python escape, e.g. {{ slice.creator is not None }}
+ lhs, rhs = v
+
+ assignments = []
+
+ try:
+ for t in lhs, rhs:
+ py_expr = t['python']
+
+ self.verdict_next()
+ vv = self.verdict_variable
+
+ try:
+ expr_ast = self.str_to_ast(py_expr)
+ except SyntaxError:
+ raise PolicyException('Syntax error in %s' % v)
+
+ if not isinstance(expr_ast, ast.Expr):
+ raise PolicyException(
+ '%s is not an expression' % expr_ast)
+
+ assignment_str = """
+%(verdict_var)s = (%(escape_expr)s)
+ """ % {'verdict_var': vv, 'escape_expr': py_expr}
+
+ if t == lhs:
+ lhs = vv
+ else:
+ rhs = vv
+
+ assignment_ast = self.str_to_ast(assignment_str)
+ assignments.append(assignment_ast)
+ except TypeError:
+ pass
+
+ comparison_str = """
+%(verdict_var)s = (%(lhs)s == %(rhs)s)
+ """ % {'verdict_var': verdict_var, 'lhs': lhs, 'rhs': rhs}
+
+ comparison_ast = self.str_to_ast(comparison_str)
+
+ combined_ast = ast.Module(body=assignments + [comparison_ast])
+ return combined_ast
+ elif k in BINOPS:
+ lhs, rhs = v
+
+ top_vvar = self.verdict_variable
+
+ self.verdict_next()
+ lvar = self.verdict_variable
+
+ self.verdict_next()
+ rvar = self.verdict_variable
+
+ lblock = self.gen_test(lhs, lvar)
+ rblock = self.gen_test(rhs, rvar)
+
+ invert = ''
+ if k == '&':
+ binop = 'and'
+ elif k == '|':
+ binop = 'or'
+ elif k == '=>':
+ binop = 'or'
+ invert = 'not'
+
+ binop_str = """
+%(verdict_var)s = %(invert)s %(lvar)s %(binop)s %(rvar)s
+ """ % {'verdict_var': top_vvar, 'invert': invert, 'lvar': lvar, 'binop': binop, 'rvar': rvar}
+
+ binop_ast = self.str_to_ast(binop_str)
+
+ combined_ast = ast.Module(body=[lblock, rblock, binop_ast])
+ return combined_ast
+ elif k == 'exists':
+ # If the variable starts with a capital letter,
+ # we assume that it is a model. If it starts with
+ # a small letter, we assume it is an enumerable
+ #
+ # We do not support nested exists yet. FIXME.
+
+ var, expr = v
+
+ if var.istitle():
+ f = self.fol_to_python_filter(var, expr, django=True)
+ entry = f.pop()
+
+ python_str = """
+%(verdict_var)s = %(model)s.objects.filter(%(query)s)[0]
+ """ % {'verdict_var': verdict_var, 'model': var, 'query': entry}
+
+ python_ast = ast.parse(python_str)
+ else:
+ f = self.fol_to_python_filter(var, expr, django=False)
+ entry = f.pop()
+
+ python_str = """
+%(verdict_var)s = filter(lambda __elt:%(query)s, %(model)s)
+ """ % {'verdict_var': verdict_var, 'model': var, 'query': entry}
+
+ python_ast = ast.parse(python_str)
+
+ return python_ast
+ elif k=='forall':
+ var, expr = v
+
+ if var.istitle():
+ f = self.fol_to_python_filter(var, expr, django=True, negate = True)
+ entry = f.pop()
+
+ self.verdict_next()
+ vvar = self.verdict_variable
+
+ python_str = """
+%(verdict_var)s = %(model)s.objects.filter(%(query)s)[0]
+ """ % {'verdict_var': vvar, 'model': var, 'query': entry}
+
+ python_ast = ast.parse(python_str)
+ else:
+ f = self.fol_to_python_filter(var, expr, django=False, negate = True)
+ entry = f.pop()
+
+ python_str = """
+%(verdict_var)s = next(elt for elt in %(model)s if %(query)s)
+ """ % {'verdict_var': vvar, 'model': var, 'query': entry}
+
+ python_ast = ast.parse(python_str)
+
+ negate_str = """
+%(verdict_var)s = not %(vvar)s
+ """ % {'verdict_var': verdict_var, 'vvar': vvar}
+
+ negate_ast = ast.parse(negate_str)
+
+ return ast.Module(body=[python_ast, negate_ast])
+
+def xproto_fol_to_python_test(fol, model, tag=None):
+ f2p = FOL2Python()
+ fol = f2p.hoist_constants(fol)
+ a = f2p.gen_test_function(fol, 'output', tag)
+ return a
+
+
+def main():
+ while True:
+ inp = raw_input()
+ fol_lexer = lex.lex(module=FOLLexer())
+ fol_parser = yacc.yacc(module=FOLParser(), start='goal')
+
+ val = fol_parser.parse(inp, lexer=fol_lexer)
+ a = xproto_fol_to_python_test(val, 'output')
+ print a
+
+
+if __name__ == "__main__":
+ main()
diff --git a/lib/xos-genx/xosgenx/xos2jinja.py b/lib/xos-genx/xosgenx/xos2jinja.py
index c58260e..edd80ef 100644
--- a/lib/xos-genx/xosgenx/xos2jinja.py
+++ b/lib/xos-genx/xosgenx/xos2jinja.py
@@ -124,6 +124,16 @@
return True
+ def visit_PolicyDefinition(self, obj):
+ if self.package:
+ pname = '.'.join([self.package, obj.name.value.pval])
+ else:
+ pname = obj.name.value.pval
+
+ self.policies[pname] = obj.body
+
+ return True
+
def visit_PackageStatement(self, obj):
dotlist = obj.name.value
dotlist2 = [f.pval for f in dotlist]