Writing HTTP check rules

oslo.policy has supported the following syntax for a while:

http:<target URL>, which delegates the check to a remote server

Starting with 1.29, oslo.policy will also support https url(s) as well:

https:<target URL>, which delegates the check to a remote server

Both http and https support are implemented as custom check rules. If you see the setup.cfg for oslo.policy, you can see the following entry points:

oslo.policy.rule_checks =
    http = oslo_policy._external:HttpCheck
    https = oslo_policy._external:HttpsCheck

When a policy is evaluated, when the engine encounters https like in a snippet below:

       "target 1" : "https://foo.bar/baz",

The engine will look for a plugin named https in the rule_checks entry point and will try to invoke that stevedore plugin.

This mechanism allows anyone to write their own code, in their own library with their own custom stevedore based rule check plugins and can enhance their policies with custom checks. This would be useful for example to integrate with an in-house policy server.

Example code - HttpCheck


Full source located at _external.py

 1class HttpCheck(_checks.Check):
 2    """Check ``http:`` rules by calling to a remote server.
 4    This example implementation simply verifies that the response
 5    is exactly ``True``.
 6    """
 8    def __call__(self, target, creds, enforcer, current_rule=None):
 9        url = ('http:' + self.match) % target
10        data, json = self._construct_payload(creds, current_rule,
11                                             enforcer, target)
12        with contextlib.closing(
13                requests.post(url, json=json, data=data)
14        ) as r:
15            return r.text.lstrip('"').rstrip('"') == 'True'
17    @staticmethod
18    def _construct_payload(creds, current_rule, enforcer, target):
19        # Convert instances of object() in target temporarily to
20        # empty dict to avoid circular reference detection
21        # errors in jsonutils.dumps().
22        temp_target = copy.deepcopy(target)
23        for key in target.keys():
24            element = target.get(key)
25            if type(element) is object:
26                temp_target[key] = {}
27        data = json = None
28        if (enforcer.conf.oslo_policy.remote_content_type ==
29                'application/x-www-form-urlencoded'):
30            data = {'rule': jsonutils.dumps(current_rule),
31                    'target': jsonutils.dumps(temp_target),
32                    'credentials': jsonutils.dumps(creds)}
33        else:
34            json = {'rule': current_rule,
35                    'target': temp_target,
36                    'credentials': creds}
37        return data, json