Writing custom 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 a in-house policy server.
Example code - HttpCheck¶
Note
Full source located at _external.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 | 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):
url = ('http:' + self.match) % target
data, json = self._construct_payload(creds, current_rule,
enforcer, target)
with contextlib.closing(
requests.post(url, json=json, data=data)
) as r:
return r.text.lstrip('"').rstrip('"') == 'True'
@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)}
else:
json = {'rule': current_rule,
'target': temp_target,
'credentials': creds}
return data, json
|
Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.