Coverage for gws-app/gws/plugin/account/helper.py: 0%
203 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
1"""Account helper."""
3from typing import Optional, cast
5import gws
6import gws.base.edit.helper
7import gws.config.util
8import gws.plugin.email_helper
9import gws.lib.image
10import gws.lib.net
11import gws.lib.otp
13from . import core
16class MfaConfig(gws.Config):
17 """Multi-factor authentication method offered to users."""
19 mfaUid: str
20 """UID of the multi-factor authentication adapter."""
21 title: str
22 """Title of the method shown to users."""
25@gws.ext.config.helper('account')
26class Config(gws.Config):
27 """Accounts table, onboarding and password settings, shared by account components."""
29 adminModel: gws.ext.config.model
30 """Model of the accounts table for account administrators."""
31 userModel: Optional[gws.ext.config.model]
32 """Model for account data that users can edit themselves."""
33 templates: list[gws.ext.config.template]
34 """Templates for account emails."""
36 usernameColumn: str = 'email'
37 """Column of the accounts table used as the login name."""
39 passwordCreateSql: Optional[str]
40 """SQL expression for computing password hashes."""
41 passwordVerifySql: Optional[str]
42 """SQL expression for checking a password against the stored hash."""
44 tcLifeTime: gws.Duration = '3600'
45 """Validity period of temporary codes."""
47 mfa: Optional[list[MfaConfig]]
48 """Multi-factor authentication methods the user can choose from."""
49 mfaIssuer: str = ''
50 """Issuer name for multi-factor key URIs (QR codes)."""
52 onboardingUrl: str
53 """Onboarding page URL sent to new users."""
54 onboardingCompletionUrl: str = ''
55 """URL to redirect to after onboarding."""
58##
61class Error(gws.Error):
62 """Account-related error."""
64 pass
67##
70class MfaOption(gws.Data):
71 """Configured MFA method."""
73 index: int
74 """Position in the ``mfa`` configuration list, starting with 1."""
75 title: str
76 """Title shown to users."""
77 adapter: Optional[gws.AuthMultiFactorAdapter]
78 """MFA adapter, ``None`` for the "no MFA" option."""
81_DEFAULT_PASSWORD_CREATE_SQL = "crypt( {password}, gen_salt('bf') )"
82_DEFAULT_PASSWORD_VERIFY_SQL = 'crypt( {password}, {passwordColumn} )'
85@gws.ext.object.helper('account')
86class Object(gws.base.edit.helper.Object):
87 """Account helper.
89 Holds the account configuration and implements account queries and updates in the table of ``adminModel``,
90 temporary codes, passwords, MFA and emails. As an edit helper, it serves the edit API
91 of the ``accountadmin`` action for ``adminModel``.
92 """
94 adminModel: gws.DatabaseModel
95 """Model of the accounts table for administrators."""
96 userModel: gws.DatabaseModel
97 """Model for account data that users can edit themselves, not created yet."""
98 templates: list[gws.Template]
99 """Email templates."""
101 mfaIssuer: str
102 """Issuer name for MFA key URIs."""
103 mfaOptions: list[MfaOption]
104 """Configured MFA methods."""
105 onboardingUrl: str
106 """Onboarding page URL sent to new users."""
107 onboardingCompletionUrl: str
108 """URL to go to after the onboarding, defaults to ``onboardingUrl``."""
109 passwordCreateSql: str
110 """SQL expression that computes a password hash, with the placeholders ``{password}`` and ``{passwordColumn}``."""
111 passwordVerifySql: str
112 """SQL expression that computes the hash to compare with the stored one, with the same placeholders."""
113 tcLifeTime: int
114 """Validity period of temporary codes in seconds."""
115 usernameColumn: str
116 """Column with the login name."""
118 def configure(self):
119 self.configure_templates()
121 self.adminModel = cast(gws.DatabaseModel, self.create_child(gws.ext.object.model, self.cfg('adminModel')))
123 self.mfaIssuer = self.cfg('mfaIssuer')
124 self.mfaOptions = []
126 self.onboardingUrl = self.cfg('onboardingUrl')
127 self.onboardingCompletionUrl = self.cfg('onboardingCompletionUrl') or self.onboardingUrl
129 self.passwordCreateSql = self.cfg('passwordCreateSql', default=_DEFAULT_PASSWORD_CREATE_SQL)
130 self.passwordVerifySql = self.cfg('passwordVerifySql', default=_DEFAULT_PASSWORD_VERIFY_SQL)
132 self.tcLifeTime = self.cfg('tcLifeTime', default=3600)
134 self.usernameColumn = self.cfg('usernameColumn', default=core.Columns.email)
136 def configure_templates(self):
137 """Configure the email templates.
139 Returns:
140 ``True`` if templates were configured.
141 """
144 return gws.config.util.configure_templates_for(self)
146 def post_configure(self):
147 for n, c in enumerate(self.cfg('mfa', default=[]), 1):
148 opt = MfaOption(index=n, title=c.title)
149 if c.mfaUid:
150 opt.adapter = self.root.get(c.mfaUid)
151 if not opt.adapter:
152 raise gws.ConfigurationError(f'MFA Adapter not found {c.mfaUid=}')
153 self.mfaOptions.append(opt)
155 ##
157 def get_models(self, req, p):
158 return [self.adminModel]
160 def write_feature(self, req, p):
161 is_new = p.feature.isNew
162 f = super().write_feature(req, p)
164 if f and not f.errors and is_new:
165 account = self.get_account_by_id(f.uid())
166 self.reset(account)
168 return f
170 ##
172 def get_account_by_id(self, uid: str) -> Optional[dict]:
173 """Find an account by its primary key.
175 Args:
176 uid: Primary key value.
178 Returns:
179 The account record, or ``None`` if not found.
180 """
183 sql = f"""
184 SELECT * FROM {self.adminModel.tableName}
185 WHERE {self.adminModel.uidName}=:uid
186 """
187 rs = self.adminModel.db.select_text(sql, uid=uid)
188 return rs[0] if rs else None
190 def get_account_by_credentials(self, credentials: gws.Data, expected_status: Optional[core.Status] = None) -> Optional[dict]:
191 """Find an account by login name and password.
193 Args:
194 credentials: Credentials with ``username`` and ``password``.
195 expected_status: If given, the account must have this status.
197 Returns:
198 The account record, or ``None`` if the login name is not found.
200 Raises:
201 Error: If the login name is not unique, the password is wrong, or the status is not the expected one.
202 """
205 expr = self.passwordVerifySql
206 expr = expr.replace('{password}', ':password')
207 expr = expr.replace('{passwordColumn}', core.Columns.password)
209 username = credentials.get('username')
210 password = credentials.get('password')
212 sql = f"""
213 SELECT
214 {self.adminModel.uidName},
215 ( {core.Columns.password} = {expr} ) AS validpassword,
216 {core.Columns.status}
217 FROM
218 {self.adminModel.tableName}
219 WHERE
220 {self.usernameColumn} = :username
221 """
222 rs = self.adminModel.db.select_text(sql, username=username, password=password)
224 if not rs:
225 gws.log.warning(f'get_account_by_credentials: {username=} not found')
226 return
228 if len(rs) > 1:
229 raise Error(f'get_account_by_credentials: multiple entries for {username=}')
231 r = rs[0]
233 if not r.get('validpassword'):
234 raise Error(f'get_account_by_credentials: {username=} wrong password')
236 if expected_status:
237 status = r.get(core.Columns.status)
238 if status != expected_status:
239 raise Error(f'get_account_by_credentials: {username=} wrong {status=} {expected_status=}')
241 return self.get_account_by_id(self.get_uid(r))
243 def get_account_by_tc(self, tc: str, category: str, expected_status: Optional[core.Status] = None) -> Optional[dict]:
244 """Find an account by a temporary code.
246 If the code is found, it is invalidated, so that it can be used only once.
248 Args:
249 tc: Temporary code.
250 category: Expected category of the code.
251 expected_status: If given, the account must have this status.
253 Returns:
254 The account record, or ``None`` if the code is not found, has a different category, is expired,
255 or the account has a different status.
257 Raises:
258 Error: If the code is not unique.
259 """
262 sql = f"""
263 SELECT
264 {self.adminModel.uidName},
265 {core.Columns.tcTime},
266 {core.Columns.tcCategory},
267 {core.Columns.status}
268 FROM
269 {self.adminModel.tableName}
270 WHERE
271 {core.Columns.tc} = :tc
272 """
273 rs = self.adminModel.db.select_text(sql, tc=tc)
275 if not rs:
276 gws.log.warning(f'get_account_by_tc: {tc=} not found')
277 return
279 self.invalidate_tc(tc)
281 if len(rs) > 1:
282 raise Error(f'get_account_by_tc: {tc=} multiple entries')
284 r = rs[0]
286 if r.get(core.Columns.tcCategory) != category:
287 gws.log.warning(f'get_account_by_tc: {category=} {tc=} wrong category')
288 return
290 if gws.u.stime() - r.get(core.Columns.tcTime, 0) > self.tcLifeTime:
291 gws.log.warning(f'get_account_by_tc: {category=} {tc=} expired')
292 return
294 if expected_status:
295 status = r.get(core.Columns.status)
296 if status != expected_status:
297 gws.log.warning(f'get_account_by_tc: {category=} {tc=} wrong {status=} {expected_status=}')
298 return
300 return self.get_account_by_id(self.get_uid(r))
302 ##
304 def set_password(self, account: dict, password):
305 """Store the hash of a new password.
307 Args:
308 account: Account record.
309 password: New password in plain text.
310 """
313 expr = self.passwordCreateSql
314 expr = expr.replace('{password}', ':password')
315 expr = expr.replace('{passwordColumn}', core.Columns.password)
317 sql = f"""
318 UPDATE {self.adminModel.tableName}
319 SET
320 {core.Columns.password} = {expr}
321 WHERE
322 {self.adminModel.uidName} = :uid
323 """
324 self.adminModel.db.execute_text(sql, password=password, uid=self.get_uid(account))
326 def validate_password(self, password: str) -> bool:
327 """Check if a password is acceptable.
329 Currently only empty passwords are rejected.
331 Args:
332 password: Password in plain text.
334 Returns:
335 ``True`` if the password is acceptable.
336 """
339 if len(password.strip()) == 0:
340 return False
341 # @TODO password complexity validation
342 return True
344 ##
346 def set_mfa(self, account: dict, mfa_option_index: int):
347 """Store the selected MFA method of an account.
349 Args:
350 account: Account record.
351 mfa_option_index: Index of the MFA option.
353 Raises:
354 Error: If there is no option with this index.
355 """
358 mfa_uid = None
360 for mo in self.mfa_options(account):
361 if mo.index == mfa_option_index:
362 mfa_uid = mo.adapter.uid if mo.adapter else ''
363 break
365 if mfa_uid is None:
366 raise Error(f'{mfa_option_index=} not found')
368 sql = f"""
369 UPDATE {self.adminModel.tableName}
370 SET
371 {core.Columns.mfaUid} = :mfa_uid
372 WHERE
373 {self.adminModel.uidName} = :uid
374 """
375 self.adminModel.db.execute_text(sql, mfa_uid=mfa_uid, uid=self.get_uid(account))
377 def mfa_options(self, account: dict) -> list[MfaOption]:
378 """Get the MFA methods available to an account.
380 Currently all configured methods are available to all accounts.
382 Args:
383 account: Account record.
385 Returns:
386 A list of MFA options.
387 """
390 # @TODO different options per account
391 return self.mfaOptions
393 def generate_mfa_secret(self, account: dict) -> str:
394 """Generate and store a new MFA secret.
396 Args:
397 account: Account record.
399 Returns:
400 The secret.
401 """
404 secret = gws.lib.otp.random_secret()
406 sql = f"""
407 UPDATE {self.adminModel.tableName}
408 SET
409 {core.Columns.mfaSecret} = :secret
410 WHERE
411 {self.adminModel.uidName} = :uid
412 """
413 self.adminModel.db.execute_text(sql, secret=secret, uid=self.get_uid(account))
415 return secret
417 def qr_code_for_mfa(self, account: dict, mo: MfaOption, secret: str) -> str:
418 """Create a QR code of the key URI for an MFA method.
420 Args:
421 account: Account record.
422 mo: MFA option.
423 secret: MFA secret.
425 Returns:
426 The QR code image as a data URL, or an empty string if the method has no adapter or no key URI.
427 """
430 if not mo.adapter:
431 return ''
432 url = mo.adapter.key_uri(secret, self.mfaIssuer, account.get(self.usernameColumn))
433 if not url:
434 return ''
435 return gws.lib.image.qr_code(url).to_data_url()
437 ##
439 def set_status(self, account: dict, status: core.Status):
440 """Set the status of an account.
442 Args:
443 account: Account record.
444 status: New status.
445 """
448 sql = f"""
449 UPDATE {self.adminModel.tableName}
450 SET
451 {core.Columns.status} = :status
452 WHERE
453 {self.adminModel.uidName} = :uid
454 """
455 self.adminModel.db.execute_text(sql, status=status, uid=self.get_uid(account))
457 def reset(self, account: dict):
458 """Reset an account.
460 Clears the password and the MFA secret and sets the status to ``new``.
461 If ``onboardingUrl`` is configured, the onboarding email is sent.
463 Args:
464 account: Account record.
466 Raises:
467 Error: If the onboarding email is due and the account has no email address.
468 """
471 sql = f"""
472 UPDATE {self.adminModel.tableName}
473 SET
474 {core.Columns.status} = :status,
475 {core.Columns.password} = '',
476 {core.Columns.mfaSecret} = ''
477 WHERE
478 {self.adminModel.uidName} = :uid
479 """
480 self.adminModel.db.execute_text(sql, status=core.Status.new, uid=self.get_uid(account))
482 if self.onboardingUrl:
483 self.send_onboarding_email(account)
485 ##
487 def send_onboarding_email(self, account: dict):
488 """Generate an onboarding code and send the onboarding email with the link.
490 Args:
491 account: Account record.
493 Raises:
494 Error: If the account has no email address.
495 """
498 tc = self.generate_tc(account, core.Category.onboarding)
499 url = gws.lib.net.add_params(self.onboardingUrl, onboarding=tc)
500 self.send_mail(account, core.Category.onboarding, {'url': url})
502 def generate_tc(self, account: dict, category: str) -> str:
503 """Generate and store a new temporary code.
505 Args:
506 account: Account record.
507 category: Code category.
509 Returns:
510 The code.
511 """
514 tc = self.make_tc()
516 sql = f"""
517 UPDATE {self.adminModel.tableName}
518 SET
519 {core.Columns.tc} = :tc,
520 {core.Columns.tcTime} = :time,
521 {core.Columns.tcCategory} = :category
522 WHERE
523 {self.adminModel.uidName} = :uid
524 """
525 self.adminModel.db.execute_text(sql, tc=tc, time=gws.u.stime(), category=category, uid=self.get_uid(account))
527 return tc
529 def clear_tc(self, account: dict):
530 """Remove the temporary code of an account.
532 Args:
533 account: Account record.
534 """
537 sql = f"""
538 UPDATE {self.adminModel.tableName}
539 SET
540 {core.Columns.tc} = '',
541 {core.Columns.tcTime} = 0,
542 {core.Columns.tcCategory} = ''
543 WHERE
544 {self.adminModel.uidName} = :uid
545 """
546 self.adminModel.db.execute_text(sql, uid=self.get_uid(account))
548 def invalidate_tc(self, tc: str):
549 """Remove a temporary code from all accounts that have it.
551 Args:
552 tc: Temporary code.
553 """
556 sql = f"""
557 UPDATE {self.adminModel.tableName}
558 SET
559 {core.Columns.tc} = '',
560 {core.Columns.tcTime} = 0,
561 {core.Columns.tcCategory} = ''
562 WHERE
563 {core.Columns.tc} = :tc
564 """
565 self.adminModel.db.execute_text(sql, tc=tc)
567 ##
569 def get_uid(self, account: dict) -> str:
570 """Get the primary key of an account.
572 Args:
573 account: Account record.
575 Returns:
576 The primary key value.
577 """
580 return account.get(self.adminModel.uidName)
582 def make_tc(self):
583 """Create a random temporary code.
585 Returns:
586 A random string of 32 characters.
587 """
590 return gws.u.random_string(32)
592 def send_mail(self, account: dict, category: str, args: Optional[dict] = None):
593 """Send an email to an account.
595 The subject and the body are rendered from the templates ``<category>.emailSubject`` and
596 ``<category>.emailBody`` (plain text and HTML), and sent with the ``email`` helper.
598 Args:
599 account: Account record.
600 category: Email category, see ``core.Category``.
601 args: Template arguments, the account record is added as ``account``.
603 Raises:
604 Error: If the account has no email address.
605 """
608 email = account.get(core.Columns.email)
609 if not email:
610 raise Error(f'account {self.get_uid(account)}: no email')
612 args = args or {}
613 args['account'] = account
615 message = gws.plugin.email_helper.Message(
616 subject=self.render_template(f'{category}.emailSubject', args),
617 mailTo=email,
618 text=self.render_template(f'{category}.emailBody', args, mime_type='text/plain'),
619 html=self.render_template(f'{category}.emailBody', args, mime_type='text/html'),
620 )
622 email_helper = cast(gws.plugin.email_helper.Object, self.root.app.helper('email'))
623 email_helper.send_mail(message)
625 def render_template(self, subject, args, mime_type=None):
626 """Render a template of this helper.
628 Args:
629 subject: Template subject.
630 args: Template arguments.
631 mime_type: Template mime type.
633 Returns:
634 The rendered content, or an empty string if there is no such template.
635 """
638 tpl = self.root.app.templateMgr.find_template(subject, where=[self], mime_type=mime_type)
639 if tpl:
640 res = tpl.render(gws.TemplateRenderInput(args=args))
641 return res.content
642 return ''