Coverage for gws-app/gws/base/auth/user.py: 84%

1"""User objects and conversion of provider records to users."""

3from typing import Optional, cast

5import gws

6import gws.lib.jsonx

9class Props(gws.Props):

10 displayName: str

11 attributes: dict

14_FIELDS = {

15 'authToken',

16 'displayName',

17 'email',

18 'localUid',

19 'loginName',

20 'mfaSecret',

21 'mfaUid',

22}

25class User(gws.User):

26 """Base user.

28 Holds the user fields, roles and attributes, and implements the permission

29 checks. Subclasses for special users override the permission decision.

30 """

32 isGuest = False

34 def __init__(self, provider, roles):

35 """Create a user with empty attributes.

37 Args:

38 provider: The authentication provider of the user.

39 roles: User roles.

40 """

41 super().__init__()

43 self.authProvider = provider

45 self.attributes = {}

46 self.data = {}

47 self.roles = roles

48 self.uid = ''

50 for f in _FIELDS:

51 setattr(self, f, '')

53 def props(self, user):

54 return Props(displayName=self.displayName, attributes=self.attributes)

56 def has_role(self, role):

57 return role in self.roles

59 def can_use(self, obj, *context):

60 return self.can(gws.Access.read, obj, *context)

62 def can_read(self, obj, *context):

63 return self.can(gws.Access.read, obj, *context)

65 def can_write(self, obj, *context):

66 return self.can(gws.Access.write, obj, *context)

68 def can_create(self, obj, *context):

69 return self.can(gws.Access.create, obj, *context)

71 def can_edit(self, obj, *context):

72 return (

73 self.can(gws.Access.write, obj, *context)

74 or self.can(gws.Access.create, obj, *context)

75 or self.can(gws.Access.delete, obj, *context)

76 )

78 def can_delete(self, obj, *context):

79 return self.can(gws.Access.delete, obj, *context)

81 def can(self, access, obj, *context):

82 ci = 0

83 clen = len(context)

85 while obj:

86 bit = self.acl_bit(access, obj)

87 if bit is not None:

88 return bit == gws.c.ALLOW

89 obj = context[ci] if ci < clen else getattr(obj, 'parent', None)

90 ci += 1

92 return False

94 def acl_bit(self, access, obj):

95 if obj is self and access == gws.Access.read:

96 return gws.c.ALLOW

97 acl = obj.permissions.get(access)

98 if acl:

99 for bit, role in acl:

100 if role in self.roles:

101 return bit

102

103 def require(self, uid=None, classref=None, access=None):

104 access = access or gws.Access.read

105 obj = self.authProvider.root.get(uid, classref)

106 if not obj:

107 raise gws.NotFoundError(f'required object {classref} {uid} not found')

108 if not self.can(access, obj):

109 raise gws.ForbiddenError(f'required object {classref} {uid} forbidden')

110 return obj

111

112 def acquire(self, uid=None, classref=None, access=None):

113 access = access or gws.Access.read

114 obj = self.authProvider.root.get(uid, classref)

115 if obj and self.can(access, obj):

116 return obj

117

118 def require_project(self, uid=None):

119 return cast(gws.Project, self.require(uid, gws.ext.object.project))

120

121 def require_layer(self, uid=None):

122 return cast(gws.Layer, self.require(uid, gws.ext.object.layer))

123

124

125class GuestUser(User):

126 """Guest user, used for requests without a login."""

127

128 isGuest = True

129

130

131class SystemUser(User):

132 """System user, allowed everything."""

133

134 def acl_bit(self, access, obj):

135 return gws.c.ALLOW

136

137

138class NobodyUser(User):

139 """User that is denied everything."""

140

141 def acl_bit(self, access, obj):

142 return gws.c.DENY

143

144

145class AuthorizedUser(User):

146 """Logged-in user."""

147

148 pass

149

150

151class AdminUser(User):

152 """Logged-in user with the ``admin`` role, allowed everything."""

153

154 def acl_bit(self, access, obj):

155 return gws.c.ALLOW

158##

161##

163def to_dict(usr) -> dict:

164 """Convert a user to a dict.

165

166 Args:

167 usr: The user.

168

169 Returns:

170 A dict with the user fields, ``attributes``, ``data``, ``roles`` and ``uid``.

171 """

172 d = {}

173

174 d['attributes'] = usr.attributes or {}

175 d['data'] = usr.data or {}

176 d['roles'] = list(usr.roles)

177 d['uid'] = usr.uid

178

179 for f in _FIELDS:

180 d[f] = getattr(usr, f, '')

181

182 return d

183

184

185def from_dict(provider: gws.AuthProvider, d: dict) -> gws.User:

186 """Restore a user from a dict created by ``to_dict``.

187

188 Returns the guest user if the roles contain ``guest``, an ``AdminUser`` if they

189 contain ``admin`` and an ``AuthorizedUser`` otherwise.

190

191 Args:

192 provider: The authentication provider of the user.

193 d: The dict.

194

195 Returns:

196 The user.

197 """

198 roles = set(d.get('roles', []))

199

200 if gws.c.ROLE_GUEST in roles:

201 return provider.root.app.authMgr.guestUser

202

203 if gws.c.ROLE_ADMIN in roles:

204 usr = AdminUser(provider, roles)

205 else:

206 usr = AuthorizedUser(provider, roles)

207

208 for f in _FIELDS:

209 setattr(usr, f, d.get(f, ''))

210

211 usr.attributes = d.get('attributes', {})

212 usr.data = d.get('data', {})

213 usr.roles = roles

214 usr.uid = gws.u.join_uid(provider.uid, usr.localUid)

215

216 return usr

217

218

219def from_record(provider: gws.AuthProvider, user_rec: dict) -> gws.User:

220 """Create a user from a record returned by a provider.

221

222 A provider can return an arbitrary dict of values. The entries ``authToken``,

223 ``displayName``, ``email``, ``localUid``, ``loginName``, ``mfaSecret`` and

224 ``mfaUid`` (or their lowercase forms) are copied to the user. ``roles`` and

225 ``attributes`` are copied as well, other entries are stored in the user's

226 ``data`` dict, with common LDAP attribute aliases (for example ``cn`` and

227 ``commonName``) filled in both ways.

228

229 The role ``all`` is always added. A record with the role ``guest`` returns

230 the guest user, one with ``admin`` creates an ``AdminUser``; all other users

231 get the role ``user``. ``loginName`` and ``email`` fall back to ``login`` and

232 ``email`` in the data, ``localUid`` to the login name, ``displayName`` to the

233 login name.

234

235 Args:

236 provider: The authentication provider.

237 user_rec: The record.

238

239 Returns:

240 The user.

241

242 Raises:

243 gws.Error: If the record has neither a local uid nor a login name.

244 """

245

246 data = dict(user_rec)

247

248 roles = set(gws.u.to_list(data.pop('roles', [])))

249 roles.add(gws.c.ROLE_ALL)

250

251 if gws.c.ROLE_GUEST in roles:

252 return provider.root.app.authMgr.guestUser

253

254 if gws.c.ROLE_ADMIN in roles:

255 usr = AdminUser(provider, roles)

256 else:

257 roles.add(gws.c.ROLE_USER)

258 usr = AuthorizedUser(provider, roles)

259

260 for f in _FIELDS:

261 if f in data:

262 setattr(usr, f, data.pop(f))

263 continue

264 if f.lower() in data:

265 setattr(usr, f, data.pop(f.lower()))

266 continue

267

268 usr.attributes = data.pop('attributes', {})

269 usr.data = _process_aliases(data)

270

271 if not usr.loginName and 'login' in usr.data:

272 usr.loginName = usr.data['login']

273

274 if not usr.email and 'email' in usr.data:

275 usr.email = usr.data['email']

276

277 usr.localUid = usr.localUid or usr.loginName

278 if not usr.localUid:

279 raise gws.Error(f'missing local uid for user')

280

281 usr.displayName = usr.displayName or usr.loginName

282

283 usr.uid = gws.u.join_uid(provider.uid, usr.localUid)

284

285 return usr

286

287

288_ALIASES = [

289 # https://tools.ietf.org/html/rfc4519

290 ('c', 'countryName'),

291 ('cn', 'commonName'),

292 ('dc', 'domainComponent'),

293 ('l', 'localityName'),

294 ('o', 'organizationName'),

295 ('ou', 'organizationalUnitName'),

296 ('sn', 'surname'),

297 ('st', 'stateOrProvinceName'),

298 ('street', 'streetAddress'),

299

300 # non-standard

301 ('login', 'userPrincipalName'),

302 ('mail', 'email'),

303]

304

305

306def _process_aliases(r):

307 """Fill in LDAP attribute aliases, in both directions."""

308 for a, b in _ALIASES:

309 if a in r:

310 r[b] = r[a]

311 elif b in r:

312 r[a] = r[b]

313 return r