import logging
import dataclasses
from typing import Union
import synapse.exc as s_exc
import synapse.common as s_common
import synapse.lib.base as s_base
import synapse.lib.cache as s_cache
import synapse.lib.nexus as s_nexus
import synapse.lib.config as s_config
import synapse.lib.crypto.passwd as s_passwd
logger = logging.getLogger(__name__)
reqValidRules = s_config.getJsValidator({
'type': 'array',
'items': {
'type': 'array',
'items': [
{'type': 'boolean'},
{'type': 'array', 'items': {'type': 'string'}},
],
'minItems': 2,
'maxItems': 2,
}
})
[docs]def getShadow(passwd): # pragma: no cover
'''This API is deprecated.'''
s_common.deprecated('hiveauth.getShadow()', curv='2.110.0')
salt = s_common.guid()
hashed = s_common.guid((salt, passwd))
return (salt, hashed)
[docs]def textFromRule(rule):
text = '.'.join(rule[1])
if not rule[0]:
text = '!' + text
return text
@dataclasses.dataclass(slots=True)
class _allowedReason:
value: Union[bool | None]
default: bool = False
isadmin: bool = False
islocked: bool = False
gateiden: Union[str | None] = None
roleiden: Union[str | None] = None
rolename: Union[str | None] = None
rule: tuple = ()
@property
def mesg(self):
if self.islocked:
return 'The user is locked.'
if self.default:
return 'No matching rule found.'
if self.isadmin:
if self.gateiden:
return f'The user is an admin of auth gate {self.gateiden}.'
return 'The user is a global admin.'
if self.rule:
rt = textFromRule((self.value, self.rule))
if self.gateiden:
if self.roleiden:
m = f'Matched role rule ({rt}) for role {self.rolename} on gate {self.gateiden}.'
else:
m = f'Matched user rule ({rt}) on gate {self.gateiden}.'
else:
if self.roleiden:
m = f'Matched role rule ({rt}) for role {self.rolename}.'
else:
m = f'Matched user rule ({rt}).'
return m
return 'No matching rule found.'
[docs]class Auth(s_nexus.Pusher):
'''
Auth is a user authentication and authorization stored in a Hive. Users
correspond to separate logins with different passwords and potentially
different privileges.
Users are assigned "rules". These rules are evaluated in order until a rule
matches. Each rule is a tuple of boolean, and a rule path (a sequence of
strings). Rules that are prefixes of a privilege match, i.e. a rule
('foo',) will match ('foo', 'bar').
Roles are just collections of rules. When a user is "granted" a role those
rules are assigned to that user. Unlike in an RBAC system, users don't
explicitly assume a role; they are merely a convenience mechanism to easily
assign the same rules to multiple users.
Authgates are objects that manage their own authorization. Each
AuthGate has roles and users subkeys which contain rules specific to that
user or role for that AuthGate. The roles and users of an AuthGate,
called GateRole and GateUser respectively, contain the iden of a role or
user defined prior and rules specific to that role or user; they do not
duplicate the metadata of the role or user.
Node layout::
Auth root (passed into constructor)
├ roles
│ ├ <role iden 1>
│ ├ ...
│ └ last role
├ users
│ ├ <user iden 1>
│ ├ ...
│ └ last user
└ authgates
├ <iden 1>
│ ├ roles
│ │ ├ <role iden 1>
│ │ ├ ...
│ │ └ last role
│ └ users
│ ├ <user iden 1>
│ ├ ...
│ └ last user
├ <iden 2>
│ ├ ...
└ ... last authgate
'''
async def __anit__(self, node, nexsroot=None, seed=None, maxusers=0):
'''
Args:
node (HiveNode): The root of the persistent storage for auth
'''
# Derive an iden from the parent
iden = 'auth:' + ':'.join(node.full)
await s_nexus.Pusher.__anit__(self, iden, nexsroot=nexsroot)
self.node = node
if seed is None:
seed = s_common.guid()
self.maxusers = maxusers
self.usersbyiden = {}
self.rolesbyiden = {}
self.usersbyname = {}
self.rolesbyname = {}
self.authgates = {}
self.allrole = None
self.rootuser = None
roles = await self.node.open(('roles',))
for _, node in roles:
await self._addRoleNode(node)
users = await self.node.open(('users',))
for _, node in users:
await self._addUserNode(node)
authgates = await self.node.open(('authgates',))
for _, node in authgates:
try:
await self._addAuthGate(node)
except Exception: # pragma: no cover
logger.exception('Failure loading AuthGate')
self.allrole = await self.getRoleByName('all')
if self.allrole is None:
# initialize the role of which all users are a member
guid = s_common.guid((seed, 'auth', 'role', 'all'))
await self._addRole(guid, 'all')
self.allrole = self.role(guid)
# initialize an admin user named root
self.rootuser = await self.getUserByName('root')
if self.rootuser is None:
guid = s_common.guid((seed, 'auth', 'user', 'root'))
await self._addUser(guid, 'root')
self.rootuser = self.user(guid)
await self.rootuser.setAdmin(True, logged=False)
await self.rootuser.setLocked(False, logged=False)
async def fini():
await self.allrole.fini()
await self.rootuser.fini()
[await u.fini() for u in self.users()]
[await r.fini() for r in self.roles()]
[await a.fini() for a in self.authgates.values()]
self.onfini(fini)
[docs] def users(self):
return self.usersbyiden.values()
[docs] def roles(self):
return self.rolesbyiden.values()
[docs] def role(self, iden):
return self.rolesbyiden.get(iden)
[docs] def user(self, iden):
return self.usersbyiden.get(iden)
[docs] async def reqUser(self, iden):
user = self.user(iden)
if user is None:
mesg = f'No user with iden {iden}.'
raise s_exc.NoSuchUser(mesg=mesg, user=iden)
return user
[docs] async def reqRole(self, iden):
role = self.role(iden)
if role is None:
mesg = f'No role with iden {iden}.'
raise s_exc.NoSuchRole(mesg=mesg)
return role
[docs] async def reqUserByName(self, name):
user = await self.getUserByName(name)
if user is None:
mesg = f'No user named {name}.'
raise s_exc.NoSuchUser(mesg=mesg, username=name)
return user
[docs] async def reqUserByNameOrIden(self, name):
user = await self.getUserByName(name)
if user is not None:
return user
user = self.user(name)
if user is None:
mesg = f'No user with name or iden {name}.'
raise s_exc.NoSuchUser(mesg=mesg)
return user
[docs] async def reqRoleByName(self, name):
role = await self.getRoleByName(name)
if role is None:
mesg = f'No role named {name}.'
raise s_exc.NoSuchRole(mesg=mesg)
return role
[docs] async def getUserByName(self, name):
'''
Get a user by their username.
Args:
name (str): Name of the user to get.
Returns:
HiveUser: A Hive User. May return None if there is no user by the requested name.
'''
return self.usersbyname.get(name)
[docs] async def getUserIdenByName(self, name):
user = await self.getUserByName(name)
return None if user is None else user.iden
[docs] async def getRoleByName(self, name):
return self.rolesbyname.get(name)
async def _addUserNode(self, node):
user = await HiveUser.anit(node, self)
self.usersbyiden[user.iden] = user
self.usersbyname[user.name] = user
return user
[docs] @s_nexus.Pusher.onPushAuto('user:name')
async def setUserName(self, iden, name):
if not isinstance(name, str):
raise s_exc.BadArg(mesg='setUserName() name must be a string')
user = self.usersbyname.get(name)
if user is not None:
if user.iden == iden:
return
raise s_exc.DupUserName(name=name)
user = await self.reqUser(iden)
self.usersbyname.pop(user.name, None)
self.usersbyname[name] = user
user.name = name
await user.node.set(name)
beheld = {
'iden': iden,
'valu': name,
}
await self.feedBeholder('user:name', beheld)
[docs] @s_nexus.Pusher.onPushAuto('role:name')
async def setRoleName(self, iden, name):
if not isinstance(name, str):
raise s_exc.BadArg(mesg='setRoleName() name must be a string')
role = self.rolesbyname.get(name)
if role is not None:
if role.iden == iden:
return
raise s_exc.DupRoleName(name=name)
role = await self.reqRole(iden)
if role.name == 'all':
mesg = 'Role "all" may not be renamed.'
raise s_exc.BadArg(mesg=mesg)
self.rolesbyname.pop(role.name, None)
self.rolesbyname[name] = role
role.name = name
await role.node.set(name)
beheld = {
'iden': iden,
'valu': name,
}
await self.feedBeholder('role:name', beheld)
[docs] async def feedBeholder(self, evnt, info, gateiden=None, logged=True):
if self.nexsroot and self.nexsroot.started and logged:
behold = {
'event': evnt,
'offset': await self.nexsroot.index(),
'info': info
}
if gateiden:
gate = self.getAuthGate(gateiden)
if gate:
behold['gates'] = [gate.pack()]
await self.fire('cell:beholder', **behold)
[docs] @s_nexus.Pusher.onPushAuto('user:info')
async def setUserInfo(self, iden, name, valu, gateiden=None, logged=True, mesg=None):
user = await self.reqUser(iden)
info = user.info
if gateiden is not None:
info = await user.genGateInfo(gateiden)
if name in ('locked', 'archived') and not valu:
self.checkUserLimit()
await info.set(name, valu)
if mesg is None:
mesg = {
'iden': iden,
'name': name,
}
if name != 'passwd':
mesg['valu'] = valu
await self.feedBeholder('user:info', mesg, gateiden=gateiden, logged=logged)
# since any user info *may* effect auth
user.clearAuthCache()
[docs] @s_nexus.Pusher.onPushAuto('role:info')
async def setRoleInfo(self, iden, name, valu, gateiden=None, logged=True, mesg=None):
role = await self.reqRole(iden)
info = role.info
if gateiden is not None:
info = await role.genGateInfo(gateiden)
await info.set(name, valu)
if mesg is None:
mesg = {
'iden': iden,
'name': name,
'valu': valu,
}
await self.feedBeholder('role:info', mesg, gateiden=gateiden, logged=logged)
role.clearAuthCache()
async def _addRoleNode(self, node):
role = await HiveRole.anit(node, self)
self.rolesbyiden[role.iden] = role
self.rolesbyname[role.name] = role
return role
async def _addAuthGate(self, node):
gate = await AuthGate.anit(node, self)
self.authgates[gate.iden] = gate
return gate
[docs] async def addAuthGate(self, iden, authgatetype):
'''
Retrieve AuthGate by iden. Create if not present.
Note:
Not change distributed
Returns:
(HiveAuthGate)
'''
gate = self.getAuthGate(iden)
if gate is not None:
if gate.type != authgatetype:
raise s_exc.InconsistentStorage(mesg=f'Stored AuthGate is of type {gate.type}, not {authgatetype}')
return gate
path = self.node.full + ('authgates', iden)
node = await self.node.hive.open(path)
await self.node.hive.set(path, authgatetype)
return await self._addAuthGate(node)
[docs] async def delAuthGate(self, iden):
'''
Delete AuthGate by iden.
Note:
Not change distributed
'''
gate = self.getAuthGate(iden)
if gate is None:
raise s_exc.NoSuchAuthGate(iden=iden)
await gate.fini()
await gate.delete()
await gate.node.pop()
del self.authgates[iden]
[docs] def getAuthGate(self, iden):
return self.authgates.get(iden)
[docs] def getAuthGates(self):
return list(self.authgates.values())
[docs] def reqAuthGate(self, iden):
gate = self.authgates.get(iden)
if gate is None:
mesg = f'No auth gate found with iden: ({iden}).'
raise s_exc.NoSuchAuthGate(iden=iden, mesg=mesg)
return gate
[docs] def checkUserLimit(self):
'''
Check if we're at the specified user limit.
This should be called right before adding/unlocking/unarchiving a user.
Raises: s_exc.HitLimit if the number of active users is at the maximum.
'''
if self.maxusers == 0:
return
numusers = 0
for user in self.users():
if user.name == 'root':
continue
if user.isLocked() or user.isArchived():
continue
numusers += 1
if numusers >= self.maxusers:
mesg = f'Cell at maximum number of users ({self.maxusers}).'
raise s_exc.HitLimit(mesg=mesg)
[docs] async def addUser(self, name, passwd=None, email=None, iden=None):
'''
Add a User to the Hive.
Args:
name (str): The name of the User.
passwd (str): A optional password for the user.
email (str): A optional email for the user.
iden (str): A optional iden to use as the user iden.
Returns:
HiveUser: A Hive User.
'''
self.checkUserLimit()
if self.usersbyname.get(name) is not None:
raise s_exc.DupUserName(name=name)
if iden is None:
iden = s_common.guid()
else:
if not s_common.isguid(iden):
raise s_exc.BadArg(name='iden', arg=iden, mesg='Argument it not a valid iden.')
if self.usersbyiden.get(iden) is not None:
raise s_exc.DupIden(name=name, iden=iden,
mesg='User already exists for the iden.')
await self._push('user:add', iden, name)
user = self.user(iden)
if passwd is not None:
await user.setPasswd(passwd)
if email is not None:
await self.setUserInfo(user.iden, 'email', email)
# Everyone's a member of 'all'
await user.grant(self.allrole.iden)
return user
@s_nexus.Pusher.onPush('user:add')
async def _addUser(self, iden, name):
user = self.usersbyname.get(name)
if user is not None:
return
node = await self.node.open(('users', iden))
await node.set(name)
await self._addUserNode(node)
user = self.usersbyname.get(name)
await self.feedBeholder('user:add', user.pack())
[docs] async def addRole(self, name, iden=None):
if self.rolesbyname.get(name) is not None:
raise s_exc.DupRoleName(name=name)
if iden is None:
iden = s_common.guid()
await self._push('role:add', iden, name)
return self.role(iden)
@s_nexus.Pusher.onPush('role:add')
async def _addRole(self, iden, name):
role = self.rolesbyname.get(name)
if role is not None:
return
node = await self.node.open(('roles', iden))
await node.set(name)
await self._addRoleNode(node)
role = self.rolesbyname.get(name)
await self.feedBeholder('role:add', role.pack())
[docs] async def delUser(self, iden):
await self.reqUser(iden)
return await self._push('user:del', iden)
@s_nexus.Pusher.onPush('user:del')
async def _delUser(self, iden):
if iden == self.rootuser.iden:
mesg = 'User "root" may not be deleted.'
raise s_exc.BadArg(mesg=mesg)
user = self.user(iden)
if user is None:
return
udef = user.pack()
self.usersbyiden.pop(user.iden)
self.usersbyname.pop(user.name)
path = self.node.full + ('users', user.iden)
for gate in self.authgates.values():
await gate._delGateUser(user.iden)
await user.fini()
await self.node.hive.pop(path)
await self.fire('user:del', udef=udef)
await self.feedBeholder('user:del', {'iden': iden})
def _getUsersInRole(self, role):
for user in self.users():
if role.iden in user.info.get('roles', ()):
yield user
[docs] async def delRole(self, iden):
await self.reqRole(iden)
return await self._push('role:del', iden)
@s_nexus.Pusher.onPush('role:del')
async def _delRole(self, iden):
if iden == self.allrole.iden:
mesg = 'Role "all" may not be deleted.'
raise s_exc.BadArg(mesg=mesg)
role = self.role(iden)
if role is None:
return
for user in self._getUsersInRole(role):
await user.revoke(role.iden, nexs=False)
for gate in self.authgates.values():
await gate._delGateRole(role.iden)
self.rolesbyiden.pop(role.iden)
self.rolesbyname.pop(role.name)
await role.fini()
# directly set the node's value and let events prop
path = self.node.full + ('roles', role.iden)
await self.node.hive.pop(path)
await self.feedBeholder('role:del', {'iden': iden})
[docs]class AuthGate(s_base.Base):
'''
The storage object for object specific rules for users/roles.
'''
async def __anit__(self, node, auth):
await s_base.Base.__anit__(self)
self.auth = auth
self.iden = node.name()
self.type = node.valu
self.node = node
self.gateroles = {} # iden -> HiveRole
self.gateusers = {} # iden -> HiveUser
for useriden, usernode in await node.open(('users',)):
user = self.auth.user(useriden)
if user is None: # pragma: no cover
logger.warning(f'Hive: path {useriden} refers to unknown user')
continue
userinfo = await usernode.dict()
self.gateusers[user.iden] = user
user.authgates[self.iden] = userinfo
user.clearAuthCache()
for roleiden, rolenode in await node.open(('roles',)):
role = self.auth.role(roleiden)
if role is None: # pragma: no cover
logger.warning(f'Hive: path {roleiden} refers to unknown role')
continue
roleinfo = await rolenode.dict()
self.gateroles[role.iden] = role
role.authgates[self.iden] = roleinfo
[docs] async def genUserInfo(self, iden):
node = await self.node.open(('users', iden))
userinfo = await node.dict()
user = self.auth.user(iden)
self.gateusers[iden] = user
user.authgates[self.iden] = userinfo
return userinfo
[docs] async def genRoleInfo(self, iden):
node = await self.node.open(('roles', iden))
roleinfo = await node.dict()
role = self.auth.role(iden)
self.gateroles[iden] = role
role.authgates[self.iden] = roleinfo
return roleinfo
async def _delGateUser(self, iden):
self.gateusers.pop(iden, None)
await self.node.pop(('users', iden))
async def _delGateRole(self, iden):
self.gateroles.pop(iden, None)
await self.node.pop(('roles', iden))
[docs] async def delete(self):
await self.fini()
for _, user in list(self.gateusers.items()):
user.authgates.pop(self.iden, None)
user.clearAuthCache()
for _, role in list(self.gateroles.items()):
role.authgates.pop(self.iden, None)
role.clearAuthCache()
await self.node.pop()
[docs] def pack(self):
users = []
for user in self.gateusers.values():
gateinfo = user.authgates.get(self.iden)
if gateinfo is None:
continue
users.append({
'iden': user.iden,
'rules': gateinfo.get('rules', ()),
'admin': gateinfo.get('admin', False),
})
roles = []
for role in self.gateroles.values():
gateinfo = role.authgates.get(self.iden)
if gateinfo is None:
continue
roles.append({
'iden': role.iden,
'rules': gateinfo.get('rules', ()),
'admin': gateinfo.get('admin', False),
})
return {
'iden': self.iden,
'type': self.type,
'users': users,
'roles': roles,
}
[docs]class HiveRuler(s_base.Base):
'''
A HiveNode that holds a list of rules. This includes HiveUsers, HiveRoles, and the AuthGate variants of those
'''
async def __anit__(self, node, auth):
await s_base.Base.__anit__(self)
self.iden = node.name()
self.auth = auth
self.node = node
self.name = node.valu
self.info = await node.dict()
self.info.setdefault('admin', False)
self.info.setdefault('rules', ())
self.authgates = {}
async def _setRulrInfo(self, name, valu, gateiden=None, nexs=True, mesg=None): # pragma: no cover
raise s_exc.NoSuchImpl(mesg='Subclass must implement _setRulrInfo')
[docs] def getRules(self, gateiden=None):
if gateiden is None:
return list(self.info.get('rules', ()))
gateinfo = self.authgates.get(gateiden)
if gateinfo is None:
return []
return list(gateinfo.get('rules', ()))
[docs] async def setRules(self, rules, gateiden=None, nexs=True, mesg=None):
reqValidRules(rules)
return await self._setRulrInfo('rules', rules, gateiden=gateiden, nexs=nexs, mesg=mesg)
[docs] async def addRule(self, rule, indx=None, gateiden=None, nexs=True):
reqValidRules((rule,))
rules = self.getRules(gateiden=gateiden)
mesg = {
'name': 'rule:add',
'iden': self.iden,
'valu': rule,
}
if indx is None:
rules.append(rule)
else:
rules.insert(indx, rule)
mesg['indx'] = indx
await self.setRules(rules, gateiden=gateiden, nexs=nexs, mesg=mesg)
[docs] async def delRule(self, rule, gateiden=None):
reqValidRules((rule,))
rules = self.getRules(gateiden=gateiden)
if rule not in rules:
return False
mesg = {
'name': 'rule:del',
'iden': self.iden,
'valu': rule,
}
rules.remove(rule)
await self.setRules(rules, gateiden=gateiden, mesg=mesg)
return True
[docs]class HiveRole(HiveRuler):
'''
A role within the Hive authorization subsystem.
A role in HiveAuth exists to bundle rules together so that the same
set of rules can be applied to multiple users.
'''
[docs] def pack(self):
return {
'type': 'role',
'iden': self.iden,
'name': self.name,
'rules': self.info.get('rules'),
'authgates': {name: info.pack() for (name, info) in self.authgates.items()},
}
async def _setRulrInfo(self, name, valu, gateiden=None, nexs=True, mesg=None):
if nexs:
return await self.auth.setRoleInfo(self.iden, name, valu, gateiden=gateiden, mesg=mesg)
else:
return await self.auth._hndlsetRoleInfo(self.iden, name, valu, gateiden=gateiden, logged=nexs, mesg=mesg)
[docs] async def setName(self, name):
return await self.auth.setRoleName(self.iden, name)
[docs] def clearAuthCache(self):
for user in self.auth.users():
if user.hasRole(self.iden):
user.clearAuthCache()
[docs] async def genGateInfo(self, gateiden):
info = self.authgates.get(gateiden)
if info is None:
gate = self.auth.reqAuthGate(gateiden)
info = self.authgates[gateiden] = await gate.genRoleInfo(self.iden)
return info
[docs] def allowed(self, perm, default=None, gateiden=None):
perm = tuple(perm)
if gateiden is not None:
info = self.authgates.get(gateiden)
if info is not None:
for allow, path in info.get('rules', ()):
if perm[:len(path)] == path:
return allow
return default
# 2. check role rules
for allow, path in self.info.get('rules', ()):
if perm[:len(path)] == path:
return allow
return default
[docs]class HiveUser(HiveRuler):
'''
A user (could be human or computer) of the system within HiveAuth.
Cortex-wide rules are stored here. AuthGate-specific rules for this user are stored in an GateUser.
'''
async def __anit__(self, node, auth):
await HiveRuler.__anit__(self, node, auth)
self.info.setdefault('roles', ())
self.info.setdefault('admin', False)
self.info.setdefault('passwd', None)
self.info.setdefault('locked', False)
self.info.setdefault('archived', False)
# arbitrary profile data for application layer use
prof = await self.node.open(('profile',))
self.profile = await prof.dict(nexs=True)
# TODO: max size check / max count check?
varz = await self.node.open(('vars',))
self.vars = await varz.dict(nexs=True)
self.permcache = s_cache.FixedCache(self._allowed)
self.allowedcache = s_cache.FixedCache(self._getAllowedReason)
[docs] def pack(self, packroles=False):
roles = self.info.get('roles', ())
if packroles:
_roles = []
for r in roles:
role = self.auth.role(r)
if role is None:
logger.error(f'User {self.iden} ({self.name}) contains a missing role: {r}')
continue
_roles.append(role.pack())
roles = _roles
return {
'type': 'user',
'iden': self.iden,
'name': self.name,
'rules': self.info.get('rules', ()),
'roles': roles,
'admin': self.info.get('admin', ()),
'email': self.info.get('email'),
'locked': self.info.get('locked'),
'archived': self.info.get('archived'),
'authgates': {name: info.pack() for (name, info) in self.authgates.items()},
}
async def _setRulrInfo(self, name, valu, gateiden=None, nexs=True, mesg=None):
if nexs:
return await self.auth.setUserInfo(self.iden, name, valu, gateiden=gateiden, mesg=mesg)
else:
return await self.auth._hndlsetUserInfo(self.iden, name, valu, gateiden=gateiden, logged=nexs, mesg=mesg)
[docs] async def setName(self, name):
return await self.auth.setUserName(self.iden, name)
[docs] async def allow(self, perm):
if not self.allowed(perm):
await self.addRule((True, perm), indx=0)
[docs] def allowed(self, perm, default=None, gateiden=None):
perm = tuple(perm)
return self.permcache.get((perm, default, gateiden))
def _allowed(self, pkey):
'''
NOTE: This must remain in sync with any changes to _getAllowedReason()!
'''
perm, default, gateiden = pkey
if self.info.get('locked'):
return False
if self.info.get('admin'):
return True
# 1. check authgate user rules
if gateiden is not None:
info = self.authgates.get(gateiden)
if info is not None:
if info.get('admin'):
return True
for allow, path in info.get('rules', ()):
if perm[:len(path)] == path:
return allow
# 2. check user rules
for allow, path in self.info.get('rules', ()):
if perm[:len(path)] == path:
return allow
# 3. check authgate role rules
if gateiden is not None:
for role in self.getRoles():
info = role.authgates.get(gateiden)
if info is None:
continue
for allow, path in info.get('rules', ()):
if perm[:len(path)] == path:
return allow
# 4. check role rules
for role in self.getRoles():
for allow, path in role.info.get('rules', ()):
if perm[:len(path)] == path:
return allow
return default
[docs] def getAllowedReason(self, perm, default=None, gateiden=None):
'''
A routine which will return a tuple of (allowed, info).
'''
perm = tuple(perm)
return self.allowedcache.get((perm, default, gateiden))
def _getAllowedReason(self, pkey):
'''
NOTE: This must remain in sync with any changes to _allowed()!
'''
perm, default, gateiden = pkey
if self.info.get('locked'):
return _allowedReason(False, islocked=True)
if self.info.get('admin'):
return _allowedReason(True, isadmin=True)
# 1. check authgate user rules
if gateiden is not None:
info = self.authgates.get(gateiden)
if info is not None:
if info.get('admin'):
return _allowedReason(True, isadmin=True, gateiden=gateiden)
for allow, path in info.get('rules', ()):
if perm[:len(path)] == path:
return _allowedReason(allow, gateiden=gateiden, rule=path)
# 2. check user rules
for allow, path in self.info.get('rules', ()):
if perm[:len(path)] == path:
return _allowedReason(allow, rule=path)
# 3. check authgate role rules
if gateiden is not None:
for role in self.getRoles():
info = role.authgates.get(gateiden)
if info is None:
continue
for allow, path in info.get('rules', ()):
if perm[:len(path)] == path:
return _allowedReason(allow, gateiden=gateiden, roleiden=role.iden, rolename=role.name,
rule=path)
# 4. check role rules
for role in self.getRoles():
for allow, path in role.info.get('rules', ()):
if perm[:len(path)] == path:
return _allowedReason(allow, roleiden=role.iden, rolename=role.name, rule=path)
return _allowedReason(default, default=True)
[docs] def clearAuthCache(self):
self.permcache.clear()
self.allowedcache.clear()
[docs] async def genGateInfo(self, gateiden):
info = self.authgates.get(gateiden)
if info is None:
gate = self.auth.reqAuthGate(gateiden)
info = await gate.genUserInfo(self.iden)
return info
[docs] def confirm(self, perm, default=None, gateiden=None):
if not self.allowed(perm, default=default, gateiden=gateiden):
self.raisePermDeny(perm, gateiden=gateiden)
[docs] def raisePermDeny(self, perm, gateiden=None):
perm = '.'.join(perm)
if gateiden is None:
mesg = f'User {self.name!r} ({self.iden}) must have permission {perm}'
raise s_exc.AuthDeny(mesg=mesg, perm=perm, user=self.iden, username=self.name)
gate = self.auth.reqAuthGate(gateiden)
mesg = f'User {self.name!r} ({self.iden}) must have permission {perm} on object {gate.iden} ({gate.type}).'
raise s_exc.AuthDeny(mesg=mesg, perm=perm, user=self.iden, username=self.name)
[docs] def getRoles(self):
for iden in self.info.get('roles', ()):
role = self.auth.role(iden)
if role is None:
logger.warning(f'user {self.iden} has non-existent role: {iden}')
continue
yield role
[docs] def hasRole(self, iden):
return iden in self.info.get('roles', ())
[docs] async def grant(self, roleiden, indx=None):
role = await self.auth.reqRole(roleiden)
roles = list(self.info.get('roles'))
if role.iden in roles:
return
if indx is None:
roles.append(role.iden)
else:
roles.insert(indx, role.iden)
mesg = {'name': 'role:grant', 'iden': self.iden, 'role': role.pack()}
await self.auth.setUserInfo(self.iden, 'roles', roles, mesg=mesg)
[docs] async def setRoles(self, roleidens):
'''
Replace all the roles for a given user with a new list of roles.
Args:
roleidens (list): A list of roleidens.
Notes:
The roleiden for the "all" role must be present in the new list of roles. This replaces all existing roles
that the user has with the new roles.
Returns:
None
'''
current_roles = list(self.info.get('roles'))
roleidens = list(roleidens)
if current_roles == roleidens:
return
if self.auth.allrole.iden not in roleidens:
mesg = 'Role "all" must be in the list of roles set.'
raise s_exc.BadArg(mesg=mesg)
roles = []
for iden in roleidens:
r = await self.auth.reqRole(iden)
roles.append(r.pack())
mesg = {'name': 'role:set', 'iden': self.iden, 'roles': roles}
await self.auth.setUserInfo(self.iden, 'roles', roleidens, mesg=mesg)
[docs] async def revoke(self, iden, nexs=True):
role = await self.auth.reqRole(iden)
if role.name == 'all':
mesg = 'Role "all" may not be revoked.'
raise s_exc.BadArg(mesg=mesg)
roles = list(self.info.get('roles'))
if role.iden not in roles:
return
roles.remove(role.iden)
mesg = {'name': 'role:revoke', 'iden': self.iden, 'role': role.pack()}
if nexs:
await self.auth.setUserInfo(self.iden, 'roles', roles, mesg=mesg)
else:
await self.auth._hndlsetUserInfo(self.iden, 'roles', roles, logged=nexs, mesg=mesg)
[docs] def isLocked(self):
return self.info.get('locked')
[docs] def isAdmin(self, gateiden=None):
# being a global admin always wins
admin = self.info.get('admin', False)
if admin or gateiden is None:
return admin
gateinfo = self.authgates.get(gateiden)
if gateinfo is None:
return False
return gateinfo.get('admin', False)
[docs] def reqAdmin(self, gateiden=None, mesg=None):
if self.isAdmin(gateiden=gateiden):
return
if mesg is None:
mesg = 'This action requires global admin permissions.'
if gateiden is not None:
mesg = f'This action requires admin permissions on gate: {gateiden}'
raise s_exc.AuthDeny(mesg=mesg, user=self.iden, username=self.name)
[docs] def isArchived(self):
return self.info.get('archived')
[docs] async def setAdmin(self, admin, gateiden=None, logged=True):
if not isinstance(admin, bool):
raise s_exc.BadArg(mesg='setAdmin requires a boolean')
if logged:
await self.auth.setUserInfo(self.iden, 'admin', admin, gateiden=gateiden)
else:
await self.auth._hndlsetUserInfo(self.iden, 'admin', admin, gateiden=gateiden, logged=logged)
[docs] async def setLocked(self, locked, logged=True):
if not isinstance(locked, bool):
raise s_exc.BadArg(mesg='setLocked requires a boolean')
if logged:
await self.auth.setUserInfo(self.iden, 'locked', locked)
else:
await self.auth._hndlsetUserInfo(self.iden, 'locked', locked, logged=logged)
[docs] async def setArchived(self, archived):
if not isinstance(archived, bool):
raise s_exc.BadArg(mesg='setArchived requires a boolean')
await self.auth.setUserInfo(self.iden, 'archived', archived)
if archived:
await self.setLocked(True)
[docs] async def tryPasswd(self, passwd, nexs=True):
if self.info.get('locked', False):
return False
if passwd is None:
return False
onepass = self.info.get('onepass')
if onepass is not None:
if isinstance(onepass, dict):
shadow = onepass.get('shadow')
expires = onepass.get('expires')
if expires >= s_common.now():
if await s_passwd.checkShadowV2(passwd=passwd, shadow=shadow):
await self.auth.setUserInfo(self.iden, 'onepass', None)
logger.debug(f'Used one time password for {self.name}',
extra={'synapse': {'user': self.iden, 'username': self.name}})
return True
else:
# Backwards compatible password handling
expires, params, hashed = onepass
if expires >= s_common.now():
if s_common.guid((params, passwd)) == hashed:
await self.auth.setUserInfo(self.iden, 'onepass', None)
logger.debug(f'Used one time password for {self.name}',
extra={'synapse': {'user': self.iden, 'username': self.name}})
return True
shadow = self.info.get('passwd')
if shadow is None:
return False
if isinstance(shadow, dict):
return await s_passwd.checkShadowV2(passwd=passwd, shadow=shadow)
# Backwards compatible password handling
salt, hashed = shadow
if s_common.guid((salt, passwd)) == hashed:
logger.debug(f'Migrating password to shadowv2 format for user {self.name}',
extra={'synapse': {'user': self.iden, 'username': self.name}})
# Update user to new password hashing scheme.
await self.setPasswd(passwd=passwd, nexs=nexs)
return True
return False
[docs] async def setPasswd(self, passwd, nexs=True):
# Prevent empty string or non-string values
if passwd is None:
shadow = None
elif passwd and isinstance(passwd, str):
shadow = await s_passwd.getShadowV2(passwd=passwd)
else:
raise s_exc.BadArg(mesg='Password must be a string')
if nexs:
await self.auth.setUserInfo(self.iden, 'passwd', shadow)
else:
await self.auth._hndlsetUserInfo(self.iden, 'passwd', shadow, logged=nexs)