authentik.policies.expression.tests

  1"""evaluator tests"""
  2
  3from django.test import RequestFactory, TestCase
  4from guardian.shortcuts import get_anonymous_user
  5from rest_framework.serializers import ValidationError
  6from rest_framework.test import APITestCase
  7
  8from authentik.core.models import Application
  9from authentik.lib.generators import generate_id
 10from authentik.policies.exceptions import PolicyException
 11from authentik.policies.expression.api import ExpressionPolicySerializer
 12from authentik.policies.expression.evaluator import PolicyEvaluator
 13from authentik.policies.expression.models import ExpressionPolicy
 14from authentik.policies.models import PolicyBinding
 15from authentik.policies.process import PolicyProcess
 16from authentik.policies.types import PolicyRequest
 17
 18
 19class TestEvaluator(TestCase):
 20    """Evaluator tests"""
 21
 22    def setUp(self):
 23        factory = RequestFactory()
 24        self.http_request = factory.get("/")
 25        self.obj = Application.objects.create(
 26            name=generate_id(),
 27            slug=generate_id(),
 28        )
 29        self.request = PolicyRequest(user=get_anonymous_user())
 30        self.request.obj = self.obj
 31        self.request.http_request = self.http_request
 32
 33    def test_full(self):
 34        """Test full with Policy instance"""
 35        policy = ExpressionPolicy(name="test", expression="return 'test'")
 36        policy.save()
 37        request = PolicyRequest(get_anonymous_user())
 38        result = policy.passes(request)
 39        self.assertTrue(result.passing)
 40
 41    def test_valid(self):
 42        """test simple value expression"""
 43        template = "return True"
 44        evaluator = PolicyEvaluator("test")
 45        evaluator.set_policy_request(self.request)
 46        self.assertEqual(evaluator.evaluate(template).passing, True)
 47
 48    def test_messages(self):
 49        """test expression with message return"""
 50        template = 'ak_message("some message");return False'
 51        evaluator = PolicyEvaluator("test")
 52        evaluator.set_policy_request(self.request)
 53        result = evaluator.evaluate(template)
 54        self.assertEqual(result.passing, False)
 55        self.assertEqual(result.messages, ("some message",))
 56
 57    def test_invalid_syntax(self):
 58        """test invalid syntax"""
 59        template = ";"
 60        evaluator = PolicyEvaluator("test")
 61        evaluator.set_policy_request(self.request)
 62        with self.assertRaises(PolicyException):
 63            evaluator.evaluate(template)
 64
 65    def test_validate(self):
 66        """test validate"""
 67        template = "True"
 68        evaluator = PolicyEvaluator("test")
 69        result = evaluator.validate(template)
 70        self.assertEqual(result, True)
 71
 72    def test_validate_invalid(self):
 73        """test validate"""
 74        template = ";"
 75        evaluator = PolicyEvaluator("test")
 76        with self.assertRaises(ValidationError):
 77            evaluator.validate(template)
 78
 79    def test_execution_logging(self):
 80        """test execution_logging"""
 81        expr = ExpressionPolicy.objects.create(
 82            name=generate_id(),
 83            execution_logging=True,
 84            expression="ak_message(request.http_request.path)\nreturn True",
 85        )
 86        evaluator = PolicyEvaluator("test")
 87        evaluator.set_policy_request(self.request)
 88        proc = PolicyProcess(PolicyBinding(policy=expr), request=self.request, connection=None)
 89        res = proc.profiling_wrapper()
 90        self.assertEqual(res.messages, ("/",))
 91
 92    def test_call_policy(self):
 93        """test ak_call_policy"""
 94        expr = ExpressionPolicy.objects.create(
 95            name=generate_id(),
 96            execution_logging=True,
 97            expression="ak_message(request.http_request.path)\nreturn True",
 98        )
 99        expr2 = ExpressionPolicy.objects.create(
100            name=generate_id(),
101            execution_logging=True,
102            expression=f"""
103            ak_message(request.http_request.path)
104            res = ak_call_policy('{expr.name}')
105            ak_message(request.http_request.path)
106            for msg in res.messages:
107                ak_message(msg)
108            """,
109        )
110        proc = PolicyProcess(PolicyBinding(policy=expr2), request=self.request, connection=None)
111        res = proc.profiling_wrapper()
112        self.assertEqual(res.messages, ("/", "/", "/"))
113
114    def test_call_policy_test_like(self):
115        """test ak_call_policy without `obj` set, as if it was when testing policies"""
116        expr = ExpressionPolicy.objects.create(
117            name=generate_id(),
118            execution_logging=True,
119            expression="ak_message(request.http_request.path)\nreturn True",
120        )
121        expr2 = ExpressionPolicy.objects.create(
122            name=generate_id(),
123            execution_logging=True,
124            expression=f"""
125            ak_message(request.http_request.path)
126            res = ak_call_policy('{expr.name}')
127            ak_message(request.http_request.path)
128            for msg in res.messages:
129                ak_message(msg)
130            """,
131        )
132        self.request.obj = None
133        proc = PolicyProcess(PolicyBinding(policy=expr2), request=self.request, connection=None)
134        res = proc.profiling_wrapper()
135        self.assertEqual(res.messages, ("/", "/", "/"))
136
137
138class TestExpressionPolicyAPI(APITestCase):
139    """Test expression policy's API"""
140
141    def test_validate(self):
142        """Test ExpressionPolicy's validation"""
143        # Because the root property-mapping has no write operation, we just instantiate
144        # a serializer and test inline
145        expr = "return True"
146        self.assertEqual(ExpressionPolicySerializer().validate_expression(expr), expr)
147        with self.assertRaises(ValidationError):
148            ExpressionPolicySerializer().validate_expression("/")