编写 HTTP 检查规则

oslo.policy 已经支持以下语法一段时间了

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

从 1.29 版本开始，oslo.policy 也将支持 https URL(s)

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

Both http 和 https 的支持都以自定义检查规则的形式实现。如果您查看 oslo.policy 的 setup.cfg，您会看到以下入口点

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

当策略被评估时，当引擎遇到 https 像下面的代码片段一样

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

引擎将在 rule_checks 入口点中查找名为 https 的插件，并尝试调用该 stevedore 插件。

这种机制允许任何人编写自己的代码，在自己的库中使用自己的基于 stevedore 的规则检查插件，并可以用自定义检查来增强他们的策略。例如，这对于与内部策略服务器集成会很有用。

示例代码 - HttpCheck

注意

完整源代码位于 _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)}