openstack/oslo.policy
Code Issues Proposed changes

Merge "Fixes file access using with statements."

Browse Source
This commit is contained in:
Zuul 2019-01-11 12:13:47 +00:00 committed by Gerrit Code Review
commit 1797a36766
2 changed files with 20 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
oslo_policy/shell.py
View File

@ -53,12 +53,17 @@ def flatten(d, parent_key=''):
def tool(policy_file, access_file, apply_rule, is_admin=False, def tool(policy_file, access_file, apply_rule, is_admin=False,
target_file=None): target_file=None):
access = access_file.read() with open(access_file, "rb", 0) as a:
access = a.read()
access_data = jsonutils.loads(access)['token'] access_data = jsonutils.loads(access)['token']
access_data['roles'] = [role['name'] for role in access_data['roles']] access_data['roles'] = [role['name'] for role in access_data['roles']]
access_data['project_id'] = access_data['project']['id'] access_data['project_id'] = access_data['project']['id']
access_data['is_admin'] = is_admin access_data['is_admin'] = is_admin
policy_data = policy_file.read()
with open(policy_file, "rb", 0) as p:
policy_data = p.read()
rules = policy.Rules.load(policy_data, "default") rules = policy.Rules.load(policy_data, "default")
class Object(object): class Object(object):
@ -67,7 +72,9 @@ def tool(policy_file, access_file, apply_rule, is_admin=False,
o.rules = rules o.rules = rules
if target_file: if target_file:
target = target_file.read() with open(target_file, "rb", 0) as t:
target = t.read()
target_data = flatten(jsonutils.loads(target)) target_data = flatten(jsonutils.loads(target))
else: else:
target_data = {"project_id": access_data['project_id']} target_data = {"project_id": access_data['project_id']}
@ -112,12 +119,9 @@ def main():
conf() conf()
policy = open(conf.policy, "rb", 0)
access = open(conf.access, "rb", 0)
target = open(conf.target, "rb", 0) if conf.target else None
is_admin = conf.is_admin.lower() == "true" is_admin = conf.is_admin.lower() == "true"
tool(policy, access, conf.rule, is_admin, target) tool(conf.policy, conf.access, conf.rule, is_admin, conf.target)
if __name__ == "__main__": if __name__ == "__main__":

18
oslo_policy/tests/test_shell.py
View File

@ -46,8 +46,8 @@ class CheckerTestCase(base.PolicyBaseTestCase):
@mock.patch("oslo_policy._checks.TrueCheck.__call__") @mock.patch("oslo_policy._checks.TrueCheck.__call__")
def test_pass_rule_parameters(self, call_mock): def test_pass_rule_parameters(self, call_mock):
policy_file = open(self.get_config_file_fullname('policy.yaml'), 'r') policy_file = self.get_config_file_fullname('policy.yaml')
access_file = open(self.get_config_file_fullname('access.json'), 'r') access_file = self.get_config_file_fullname('access.json')
apply_rule = None apply_rule = None
is_admin = False is_admin = False
stdout = self._capture_stdout() stdout = self._capture_stdout()
@ -74,8 +74,8 @@ class CheckerTestCase(base.PolicyBaseTestCase):
def test_pass_rule_parameters_sorted(self): def test_pass_rule_parameters_sorted(self):
self.create_config_file("policy.yaml", self.SAMPLE_POLICY_UNSORTED) self.create_config_file("policy.yaml", self.SAMPLE_POLICY_UNSORTED)
policy_file = open(self.get_config_file_fullname('policy.yaml'), 'r') policy_file = self.get_config_file_fullname('policy.yaml')
access_file = open(self.get_config_file_fullname('access.json'), 'r') access_file = self.get_config_file_fullname('access.json')
apply_rule = None apply_rule = None
is_admin = False is_admin = False
stdout = self._capture_stdout() stdout = self._capture_stdout()
@ -114,9 +114,9 @@ passed: sampleservice:sample_rule2
"target.json", "target.json",
jsonutils.dumps(sample_target)) jsonutils.dumps(sample_target))
policy_file = open(self.get_config_file_fullname('policy.yaml'), 'r') policy_file = self.get_config_file_fullname('policy.yaml')
access_file = open(self.get_config_file_fullname('access.json'), 'r') access_file = self.get_config_file_fullname('access.json')
target_file = open(self.get_config_file_fullname('target.json'), 'r') target_file = self.get_config_file_fullname('target.json')
stdout = self._capture_stdout() stdout = self._capture_stdout()
shell.tool(policy_file, access_file, apply_rule, is_admin, shell.tool(policy_file, access_file, apply_rule, is_admin,
@ -131,8 +131,8 @@ passed: sampleservice:sample_rule2
def test_all_nonadmin(self): def test_all_nonadmin(self):
policy_file = open(self.get_config_file_fullname('policy.yaml'), 'r') policy_file = self.get_config_file_fullname('policy.yaml')
access_file = open(self.get_config_file_fullname('access.json'), 'r') access_file = self.get_config_file_fullname('access.json')
apply_rule = None apply_rule = None
is_admin = False is_admin = False
stdout = self._capture_stdout() stdout = self._capture_stdout()