particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

package list policy

Browse source
This commit is contained in:
Mike McLean 2010-03-09 12:44:20 -05:00
parent ea6c283ba5
commit 59bde48e43
4 changed files with 103 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

93
hub/kojihub.py
View file

@ -790,12 +790,14 @@ def _pkglist_add(tag_id,pkg_id,owner,block,extra_arches,event_id=None):
def pkglist_add(taginfo,pkginfo,owner=None,block=None,extra_arches=None,force=False,update=False):
"""Add to (or update) package list for tag"""
#only admins....
context.session.assertPerm('admin')
#access control comes a little later (via an assert_policy)
#should not make any changes until after policy is checked
tag = get_tag(taginfo, strict=True)
pkg = lookup_package(pkginfo, create=True)
tag_id = tag['id']
pkg_id = pkg['id']
pkg = lookup_package(pkginfo, strict=False)
if not pkg:
if not isinstance(pkginfo, basestring):
raise GenericError, "Invalid package: %s" % pkginfo
if owner is not None:
owner = get_user(owner,strict=True)['id']
action = 'add'
@ -803,13 +805,20 @@ def pkglist_add(taginfo,pkginfo,owner=None,block=None,extra_arches=None,force=Fa
action = 'update'
elif bool(block):
action = 'block'
context.session.assertLogin()
policy_data = {'tag' : tag_id, 'action' : action, 'package' : pkginfo, 'force' : force}
#don't check policy for admins using force
if not force or not context.session.hasPerm('admin'):
assert_policy('package_list', policy_data)
if not pkg:
pkg = lookup_package(pkginfo, create=True)
koji.plugin.run_callbacks('prePackageListChange', action=action, tag=tag, package=pkg, owner=owner,
block=block, extra_arches=extra_arches, force=force, update=update)
# first check to see if package is:
# already present (via inheritance)
# blocked
pkglist = readPackageList(tag_id, pkgID=pkg_id, inherit=True)
previous = pkglist.get(pkg_id,None)
pkglist = readPackageList(tag_id, pkgID=pkg['id'], inherit=True)
previous = pkglist.get(pkg['id'],None)
if previous is None:
if block is None:
block = False
@ -847,7 +856,7 @@ def pkglist_add(taginfo,pkginfo,owner=None,block=None,extra_arches=None,force=Fa
owner = context.session.user_id
else:
raise koji.GenericError, "owner not specified"
_pkglist_add(tag_id,pkg_id,owner,block,extra_arches)
_pkglist_add(tag_id, pkg['id'], owner, block, extra_arches)
koji.plugin.run_callbacks('postPackageListChange', action=action, tag=tag, package=pkg, owner=owner,
block=block, extra_arches=extra_arches, force=force, update=update)
@ -859,10 +868,13 @@ def pkglist_remove(taginfo,pkginfo,force=False):
The main reason to remove an entry like this is to remove an override so
that the package data can be inherited from elsewhere.
"""
#only admins....
context.session.assertPerm('admin')
tag = get_tag(taginfo, strict=True)
pkg = lookup_package(pkginfo, strict=True)
context.session.assertLogin()
policy_data = {'tag' : tag['id'], 'action' : 'remove', 'package' : pkg['id'], 'force' : force}
#don't check policy for admins using force
if not force or not context.session.hasPerm('admin'):
assert_policy('package_list', policy_data)
koji.plugin.run_callbacks('prePackageListChange', action='remove', tag=tag, package=pkg)
_pkglist_remove(tag['id'],pkg['id'])
koji.plugin.run_callbacks('postPackageListChange', action='remove', tag=tag, package=pkg)
@ -871,7 +883,7 @@ def pkglist_block(taginfo,pkginfo):
"""Block the package in tag"""
pkglist_add(taginfo,pkginfo,block=True)
def pkglist_unblock(taginfo,pkginfo):
def pkglist_unblock(taginfo, pkginfo, force=False):
"""Unblock the package in tag
Generally this just adds a unblocked duplicate of the blocked entry.
@ -879,6 +891,11 @@ def pkglist_unblock(taginfo,pkginfo):
the blocking entry is simply removed"""
tag = get_tag(taginfo, strict=True)
pkg = lookup_package(pkginfo, strict=True)
context.session.assertLogin()
policy_data = {'tag' : tag['id'], 'action' : 'unblock', 'package' : pkg['id'], 'force' : force}
#don't check policy for admins using force
if not force or not context.session.hasPerm('admin'):
assert_policy('package_list', policy_data)
koji.plugin.run_callbacks('prePackageListChange', action='unblock', tag=tag, package=pkg)
tag_id = tag['id']
pkg_id = pkg['id']
@ -4384,13 +4401,49 @@ class OperationTest(koji.policy.MatchTest):
name = 'operation'
field = 'operation'
def policy_get_user(data):
"""Determine user from policy data (default to logged-in user)"""
if data.has_key('user_id'):
return get_user(data['user_id'])
elif context.session.logged_in:
return get_user(context.session.user_id)
return None
def policy_get_pkg(data):
"""Determine package from policy data (default to logged-in user)
returns dict as lookup_package
if package does not exist yet, the id field will be None
"""
if data.has_key('package'):
pkginfo = lookup_package(data['package'], strict=False)
if not pkginfo:
#for some operations (e.g. adding a new package), the package
#entry may not exist yet
if isinstance(data['package'], basestring):
return {'id' : None, 'name' : data['package']}
else:
raise koji.GenericError, "Invalid package: %s" % data['package']
return pkginfo
if data.has_key('build'):
binfo = get_build(data['build'], strict=True)
return {'id' : binfo['package_id'], 'name' : binfo['name']}
#else
raise koji.GenericError, "policy requires package data"
class NewPackageTest(koji.policy.BaseSimpleTest):
"""Checks to see if a package exists yet"""
name = 'is_new_package'
def run(self, data):
return (policy_get_pkg(data)['id'] is None)
class PackageTest(koji.policy.MatchTest):
"""Checks package against glob patterns"""
name = 'package'
field = '_package'
def run(self, data):
#we need to find the package name from the base data
data[self.field] = get_build(data['build'])['name']
data[self.field] = policy_get_pkg(data)['name']
return super(PackageTest, self).run(data)
class TagTest(koji.policy.MatchTest):
@ -4502,7 +4555,9 @@ class IsBuildOwnerTest(koji.policy.BaseSimpleTest):
def run(self, data):
build = get_build(data['build'])
owner = get_user(build['owner_id'])
user = get_user(data['user_id'])
user = policy_get_user(data)
if not user:
return False
if owner['id'] == user['id']:
return True
if owner['usertype'] == koji.USERTYPES['GROUP']:
@ -4520,7 +4575,9 @@ class UserInGroupTest(koji.policy.BaseSimpleTest):
"""
name = "user_in_group"
def run(self, data):
user = get_user(data['user_id'])
user = policy_get_user(data)
if not user:
return False
groups = koji.auth.get_user_groups(user['id'])
args = self.str.split()[1:]
for group_id, group in groups.iteritems():
@ -4538,7 +4595,9 @@ class HasPermTest(koji.policy.BaseSimpleTest):
"""
name = "has_perm"
def run(self, data):
user = get_user(data['user_id'])
user = policy_get_user(data)
if not user:
return False
perms = koji.auth.get_user_perms(user['id'])
args = self.str.split()[1:]
for perm in perms:
@ -4962,8 +5021,7 @@ class RootExports(object):
# the chunk rather than the whole file. the offset indicates where
# the chunk belongs
# the special offset -1 is used to indicate the final chunk
if not context.session.logged_in:
raise koji.GenericError, 'you must be logged-in to upload a file'
context.session.assertLogin()
contents = base64.decodestring(data)
del data
# we will accept offset and size as strings to work around xmlrpc limits
@ -5216,7 +5274,6 @@ class RootExports(object):
elif pkgs[pkg_id]['blocked']:
pkg_error = "Package %s blocked in %s" % (build['name'], tag['name'])
policy_data = {'tag' : tag_id, 'build' : build_id, 'fromtag' : fromtag_id}
policy_data['user_id'] = context.session.user_id
if fromtag is None:
policy_data['operation'] = 'tag'
else:
@ -5247,7 +5304,6 @@ class RootExports(object):
tag_id = get_tag(tag, strict=True)['id']
build_id = get_build(build, strict=True)['id']
policy_data = {'tag' : None, 'build' : build_id, 'fromtag' : tag_id}
policy_data['user_id'] = context.session.user_id
policy_data['operation'] = 'untag'
try:
#don't check policy for admins using force
@ -5311,7 +5367,6 @@ class RootExports(object):
#policy check
policy_data = {'tag' : tag2, 'fromtag' : tag1, 'operation' : 'move'}
policy_data['user_id'] = context.session.user_id
#don't check policy for admins using force
if not force or not context.session.hasPerm('admin'):
for build in build_list:

4
hub/kojixmlrpc.py
View file

@ -459,6 +459,10 @@ _default_policies = {
has_perm admin :: allow
all :: deny
''',
'package_list' : '''
has_perm admin :: allow
all :: deny
'''
}
def get_policy(opts, plugins):

4
koji/auth.py
View file

@ -513,6 +513,10 @@ class Session(object):
if not self.hasPerm(name) and not self.hasPerm('admin'):
raise koji.ActionNotAllowed, "%s permission required" % name
def assertLogin(self):
if not self.logged_in:
raise koji.ActionNotAllowed, "you must be logged in for this operation"
def hasGroup(self, group_id):
if not self.logged_in:
return False

21
koji/policy.py
View file

@ -60,6 +60,27 @@ class NoneTest(FalseTest):
#alias for false
class BoolTest(BaseSimpleTest):
"""Test a field in the data as a boolean value
This test can be used as-is, or it can be subclassed to
test a specific field
Syntax:
name [field]
"""
name = 'bool'
field = None
def run(self, data):
args = self.str.split()[1:]
if self.field is None:
field = args[0]
else:
# expected when we are subclassed
field = self.field
return bool(data[field])
class MatchTest(BaseSimpleTest):
"""Matches a field in the data against glob patterns