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

Note

Full source located at _external.py

class HttpCheck(_checks.Check):
    """Check ``http:`` rules by calling to a remote server.

    This example implementation simply verifies that the response
    is exactly ``True``.
    """

    def __call__(self, target, creds, enforcer, current_rule=None):
        timeout = enforcer.conf.oslo_policy.remote_timeout

        url = ('http:' + self.match) % target
        data, json = self._construct_payload(creds, current_rule,
                                             enforcer, target)
        try:
            with contextlib.closing(
                    requests.post(url, json=json, data=data, timeout=timeout)
            ) as r:
                return r.text.lstrip('"').rstrip('"') == 'True'
        except Timeout:
            raise RuntimeError("Timeout in REST API call")

    @staticmethod
    def _construct_payload(creds, current_rule, enforcer, target):
        # Convert instances of object() in target temporarily to
        # empty dict to avoid circular reference detection
        # errors in jsonutils.dumps().
        temp_target = copy.deepcopy(target)
        for key in target.keys():
            element = target.get(key)
            if type(element) is object:
                temp_target[key] = {}
        data = json = None
        if (enforcer.conf.oslo_policy.remote_content_type ==
                'application/x-www-form-urlencoded'):
            data = {'rule': jsonutils.dumps(current_rule),
                    'target': jsonutils.dumps(temp_target),
                    'credentials': jsonutils.dumps(creds)}