Coverage for gws-app/gws/plugin/account/helper.py: 66%

1"""Account helper.""" 

2 

3from typing import Optional, cast 

4 

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 

12 

13from . import core 

14 

15 

16class MfaConfig(gws.Config): 

17 """Multi-factor authentication method offered to users.""" 

18 

19 mfaUid: str 

20 """UID of the multi-factor authentication adapter.""" 

21 title: str 

22 """Title of the method shown to users.""" 

23 

24 

25@gws.ext.config.helper('account') 

26class Config(gws.Config): 

27 """Accounts table, onboarding and password settings, shared by account components.""" 

28 

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.""" 

35 

36 usernameColumn: str = 'email' 

37 """Column of the accounts table used as the login name.""" 

38 

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.""" 

43 

44 tcLifeTime: gws.Duration = '3600' 

45 """Validity period of temporary codes.""" 

46 

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).""" 

51 

52 onboardingUrl: str 

53 """Onboarding page URL sent to new users.""" 

54 onboardingCompletionUrl: str = '' 

55 """URL to redirect to after onboarding.""" 

56 

57 

58## 

59 

60 

61class Error(gws.Error): 

62 """Account-related error.""" 

63 

64 pass 

65 

66 

67## 

68 

69 

70class MfaOption(gws.Data): 

71 """Configured MFA method.""" 

72 

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.""" 

79 

80 

81_DEFAULT_PASSWORD_CREATE_SQL = "crypt( {password}, gen_salt('bf') )" 

82_DEFAULT_PASSWORD_VERIFY_SQL = 'crypt( {password}, {passwordColumn} )' 

83 

84 

85@gws.ext.object.helper('account') 

86class Object(gws.base.edit.helper.Object): 

87 """Account helper. 

88 

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 """ 

93 

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.""" 

100 

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.""" 

117 

118 def configure(self): 

119 self.configure_templates() 

120 

121 self.adminModel = cast(gws.DatabaseModel, self.create_child(gws.ext.object.model, self.cfg('adminModel'))) 

122 

123 self.mfaIssuer = self.cfg('mfaIssuer') 

124 self.mfaOptions = [] 

125 

126 self.onboardingUrl = self.cfg('onboardingUrl') 

127 self.onboardingCompletionUrl = self.cfg('onboardingCompletionUrl') or self.onboardingUrl 

128 

129 self.passwordCreateSql = self.cfg('passwordCreateSql', default=_DEFAULT_PASSWORD_CREATE_SQL) 

130 self.passwordVerifySql = self.cfg('passwordVerifySql', default=_DEFAULT_PASSWORD_VERIFY_SQL) 

131 

132 self.tcLifeTime = self.cfg('tcLifeTime', default=3600) 

133 

134 self.usernameColumn = self.cfg('usernameColumn', default=core.Columns.email) 

135 

136 def configure_templates(self): 

137 """Configure the email templates. 

138 

139 Returns: 

140 ``True`` if templates were configured. 

141 """ 

142 

143 

144 return gws.config.util.configure_templates_for(self) 

145 

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) 

154 

155 ## 

156 

157 def get_models(self, req, p): 

158 return [self.adminModel] 

159 

160 def write_feature(self, req, p): 

161 is_new = p.feature.isNew 

162 f = super().write_feature(req, p) 

163 

164 if f and not f.errors and is_new: 

165 account = self.get_account_by_id(f.uid()) 

166 self.reset(account) 

167 

168 return f 

169 

170 ## 

171 

172 def select_text(self, sql: str, **kwargs) -> list[dict]: 

173 """Execute a textual SELECT statement on the accounts database. 

174 

175 Args: 

176 sql: SQL statement with ``:name`` bind parameters. 

177 **kwargs: Bind parameters. 

178 

179 Returns: 

180 A list of rows as dicts. 

181 """ 

182 

183 with self.adminModel.db.begin() as conn: 

184 return conn.fetch_all(sql, **kwargs) 

185 

186 def execute_text(self, sql: str, **kwargs): 

187 """Execute a textual statement on the accounts database in a transaction. 

188 

189 Args: 

190 sql: SQL statement with ``:name`` bind parameters. 

191 **kwargs: Bind parameters. 

192 """ 

193 

194 with self.adminModel.db.begin() as conn: 

195 conn.execute(sql, kwargs) 

196 

197 def get_account_by_id(self, uid: str) -> Optional[dict]: 

198 """Find an account by its primary key. 

199 

200 Args: 

201 uid: Primary key value. 

202 

203 Returns: 

204 The account record, or ``None`` if not found. 

205 """ 

206 

207 

208 sql = f""" 

209 SELECT * FROM {self.adminModel.tableName} 

210 WHERE {self.adminModel.uidName}=:uid 

211 """ 

212 rs = self.select_text(sql, uid=uid) 

213 return rs[0] if rs else None 

214 

215 def get_account_by_credentials(self, credentials: gws.Data, expected_status: Optional[core.Status] = None) -> Optional[dict]: 

216 """Find an account by login name and password. 

217 

218 Args: 

219 credentials: Credentials with ``username`` and ``password``. 

220 expected_status: If given, the account must have this status. 

221 

222 Returns: 

223 The account record, or ``None`` if the login name is not found. 

224 

225 Raises: 

226 Error: If the login name is not unique, the password is wrong, or the status is not the expected one. 

227 """ 

228 

229 

230 expr = self.passwordVerifySql 

231 expr = expr.replace('{password}', ':password') 

232 expr = expr.replace('{passwordColumn}', core.Columns.password) 

233 

234 username = credentials.get('username') 

235 password = credentials.get('password') 

236 

237 sql = f""" 

238 SELECT 

239 {self.adminModel.uidName}, 

240 ( {core.Columns.password} = {expr} ) AS validpassword, 

241 {core.Columns.status} 

242 FROM 

243 {self.adminModel.tableName} 

244 WHERE 

245 {self.usernameColumn} = :username 

246 """ 

247 rs = self.select_text(sql, username=username, password=password) 

248 

249 if not rs: 

250 gws.log.warning(f'get_account_by_credentials: {username=} not found') 

251 return 

252 

253 if len(rs) > 1: 

254 raise Error(f'get_account_by_credentials: multiple entries for {username=}') 

255 

256 r = rs[0] 

257 

258 if not r.get('validpassword'): 

259 raise Error(f'get_account_by_credentials: {username=} wrong password') 

260 

261 if expected_status is not None: 

262 status = r.get(core.Columns.status) 

263 if status != expected_status: 

264 raise Error(f'get_account_by_credentials: {username=} wrong {status=} {expected_status=}') 

265 

266 return self.get_account_by_id(self.get_uid(r)) 

267 

268 def get_account_by_tc(self, tc: str, category: str, expected_status: Optional[core.Status] = None) -> Optional[dict]: 

269 """Find an account by a temporary code. 

270 

271 If the code is found, it is invalidated, so that it can be used only once. 

272 

273 Args: 

274 tc: Temporary code. 

275 category: Expected category of the code. 

276 expected_status: If given, the account must have this status. 

277 

278 Returns: 

279 The account record, or ``None`` if the code is not found, has a different category, is expired, 

280 or the account has a different status. 

281 

282 Raises: 

283 Error: If the code is not unique. 

284 """ 

285 

286 

287 sql = f""" 

288 SELECT 

289 {self.adminModel.uidName}, 

290 {core.Columns.tcTime}, 

291 {core.Columns.tcCategory}, 

292 {core.Columns.status} 

293 FROM 

294 {self.adminModel.tableName} 

295 WHERE 

296 {core.Columns.tc} = :tc 

297 """ 

298 rs = self.select_text(sql, tc=tc) 

299 

300 if not rs: 

301 gws.log.warning(f'get_account_by_tc: {tc=} not found') 

302 return 

303 

304 self.invalidate_tc(tc) 

305 

306 if len(rs) > 1: 

307 raise Error(f'get_account_by_tc: {tc=} multiple entries') 

308 

309 r = rs[0] 

310 

311 if r.get(core.Columns.tcCategory) != category: 

312 gws.log.warning(f'get_account_by_tc: {category=} {tc=} wrong category') 

313 return 

314 

315 if gws.u.stime() - r.get(core.Columns.tcTime, 0) > self.tcLifeTime: 

316 gws.log.warning(f'get_account_by_tc: {category=} {tc=} expired') 

317 return 

318 

319 if expected_status is not None: 

320 status = r.get(core.Columns.status) 

321 if status != expected_status: 

322 gws.log.warning(f'get_account_by_tc: {category=} {tc=} wrong {status=} {expected_status=}') 

323 return 

324 

325 return self.get_account_by_id(self.get_uid(r)) 

326 

327 ## 

328 

329 def set_password(self, account: dict, password): 

330 """Store the hash of a new password. 

331 

332 Args: 

333 account: Account record. 

334 password: New password in plain text. 

335 """ 

336 

337 

338 expr = self.passwordCreateSql 

339 expr = expr.replace('{password}', ':password') 

340 expr = expr.replace('{passwordColumn}', core.Columns.password) 

341 

342 sql = f""" 

343 UPDATE {self.adminModel.tableName} 

344 SET 

345 {core.Columns.password} = {expr} 

346 WHERE 

347 {self.adminModel.uidName} = :uid 

348 """ 

349 self.execute_text(sql, password=password, uid=self.get_uid(account)) 

350 

351 def validate_password(self, password: str) -> bool: 

352 """Check if a password is acceptable. 

353 

354 Currently only empty passwords are rejected. 

355 

356 Args: 

357 password: Password in plain text. 

358 

359 Returns: 

360 ``True`` if the password is acceptable. 

361 """ 

362 

363 

364 if len(password.strip()) == 0: 

365 return False 

366 # @TODO password complexity validation 

367 return True 

368 

369 ## 

370 

371 def set_mfa(self, account: dict, mfa_option_index: int): 

372 """Store the selected MFA method of an account. 

373 

374 Args: 

375 account: Account record. 

376 mfa_option_index: Index of the MFA option. 

377 

378 Raises: 

379 Error: If there is no option with this index. 

380 """ 

381 

382 

383 mfa_uid = None 

384 

385 for mo in self.mfa_options(account): 

386 if mo.index == mfa_option_index: 

387 mfa_uid = mo.adapter.uid if mo.adapter else '' 

388 break 

389 

390 if mfa_uid is None: 

391 raise Error(f'{mfa_option_index=} not found') 

392 

393 sql = f""" 

394 UPDATE {self.adminModel.tableName} 

395 SET 

396 {core.Columns.mfaUid} = :mfa_uid 

397 WHERE 

398 {self.adminModel.uidName} = :uid 

399 """ 

400 self.execute_text(sql, mfa_uid=mfa_uid, uid=self.get_uid(account)) 

401 

402 def mfa_options(self, account: dict) -> list[MfaOption]: 

403 """Get the MFA methods available to an account. 

404 

405 Currently all configured methods are available to all accounts. 

406 

407 Args: 

408 account: Account record. 

409 

410 Returns: 

411 A list of MFA options. 

412 """ 

413 

414 

415 # @TODO different options per account 

416 return self.mfaOptions 

417 

418 def generate_mfa_secret(self, account: dict) -> str: 

419 """Generate and store a new MFA secret. 

420 

421 Args: 

422 account: Account record. 

423 

424 Returns: 

425 The secret. 

426 """ 

427 

428 

429 secret = gws.lib.otp.random_secret() 

430 

431 sql = f""" 

432 UPDATE {self.adminModel.tableName} 

433 SET 

434 {core.Columns.mfaSecret} = :secret 

435 WHERE 

436 {self.adminModel.uidName} = :uid 

437 """ 

438 self.execute_text(sql, secret=secret, uid=self.get_uid(account)) 

439 

440 return secret 

441 

442 def qr_code_for_mfa(self, account: dict, mo: MfaOption, secret: str) -> str: 

443 """Create a QR code of the key URI for an MFA method. 

444 

445 Args: 

446 account: Account record. 

447 mo: MFA option. 

448 secret: MFA secret. 

449 

450 Returns: 

451 The QR code image as a data URL, or an empty string if the method has no adapter or no key URI. 

452 """ 

453 

454 

455 if not mo.adapter: 

456 return '' 

457 url = mo.adapter.key_uri(secret, self.mfaIssuer, account.get(self.usernameColumn)) 

458 if not url: 

459 return '' 

460 return gws.lib.image.util.qr_code(url).to_data_url() 

461 

462 ## 

463 

464 def set_status(self, account: dict, status: core.Status): 

465 """Set the status of an account. 

466 

467 Args: 

468 account: Account record. 

469 status: New status. 

470 """ 

471 

472 

473 sql = f""" 

474 UPDATE {self.adminModel.tableName} 

475 SET 

476 {core.Columns.status} = :status 

477 WHERE 

478 {self.adminModel.uidName} = :uid 

479 """ 

480 self.execute_text(sql, status=status, uid=self.get_uid(account)) 

481 

482 def reset(self, account: dict): 

483 """Reset an account. 

484 

485 Clears the password and the MFA secret and sets the status to ``new``. 

486 If ``onboardingUrl`` is configured, the onboarding email is sent. 

487 

488 Args: 

489 account: Account record. 

490 

491 Raises: 

492 Error: If the onboarding email is due and the account has no email address. 

493 """ 

494 

495 

496 sql = f""" 

497 UPDATE {self.adminModel.tableName} 

498 SET 

499 {core.Columns.status} = :status, 

500 {core.Columns.password} = '', 

501 {core.Columns.mfaSecret} = '' 

502 WHERE 

503 {self.adminModel.uidName} = :uid 

504 """ 

505 self.execute_text(sql, status=core.Status.new, uid=self.get_uid(account)) 

506 

507 if self.onboardingUrl: 

508 self.send_onboarding_email(account) 

509 

510 ## 

511 

512 def send_onboarding_email(self, account: dict): 

513 """Generate an onboarding code and send the onboarding email with the link. 

514 

515 Args: 

516 account: Account record. 

517 

518 Raises: 

519 Error: If the account has no email address. 

520 """ 

521 

522 

523 tc = self.generate_tc(account, core.Category.onboarding) 

524 url = gws.lib.net.add_params(self.onboardingUrl, onboarding=tc) 

525 self.send_mail(account, core.Category.onboarding, {'url': url}) 

526 

527 def generate_tc(self, account: dict, category: str) -> str: 

528 """Generate and store a new temporary code. 

529 

530 Args: 

531 account: Account record. 

532 category: Code category. 

533 

534 Returns: 

535 The code. 

536 """ 

537 

538 

539 tc = self.make_tc() 

540 

541 sql = f""" 

542 UPDATE {self.adminModel.tableName} 

543 SET 

544 {core.Columns.tc} = :tc, 

545 {core.Columns.tcTime} = :time, 

546 {core.Columns.tcCategory} = :category 

547 WHERE 

548 {self.adminModel.uidName} = :uid 

549 """ 

550 self.execute_text(sql, tc=tc, time=gws.u.stime(), category=category, uid=self.get_uid(account)) 

551 

552 return tc 

553 

554 def clear_tc(self, account: dict): 

555 """Remove the temporary code of an account. 

556 

557 Args: 

558 account: Account record. 

559 """ 

560 

561 

562 sql = f""" 

563 UPDATE {self.adminModel.tableName} 

564 SET 

565 {core.Columns.tc} = '', 

566 {core.Columns.tcTime} = 0, 

567 {core.Columns.tcCategory} = '' 

568 WHERE 

569 {self.adminModel.uidName} = :uid 

570 """ 

571 self.execute_text(sql, uid=self.get_uid(account)) 

572 

573 def invalidate_tc(self, tc: str): 

574 """Remove a temporary code from all accounts that have it. 

575 

576 Args: 

577 tc: Temporary code. 

578 """ 

579 

580 

581 sql = f""" 

582 UPDATE {self.adminModel.tableName} 

583 SET 

584 {core.Columns.tc} = '', 

585 {core.Columns.tcTime} = 0, 

586 {core.Columns.tcCategory} = '' 

587 WHERE 

588 {core.Columns.tc} = :tc 

589 """ 

590 self.execute_text(sql, tc=tc) 

591 

592 ## 

593 

594 def get_uid(self, account: dict) -> str: 

595 """Get the primary key of an account. 

596 

597 Args: 

598 account: Account record. 

599 

600 Returns: 

601 The primary key value. 

602 """ 

603 

604 

605 return account.get(self.adminModel.uidName) 

606 

607 def make_tc(self): 

608 """Create a random temporary code. 

609 

610 Returns: 

611 A random string of 32 characters. 

612 """ 

613 

614 

615 return gws.u.random_string(32) 

616 

617 def send_mail(self, account: dict, category: str, args: Optional[dict] = None): 

618 """Send an email to an account. 

619 

620 The subject and the body are rendered from the templates ``<category>.emailSubject`` and 

621 ``<category>.emailBody`` (plain text and HTML), and sent with the ``email`` helper. 

622 

623 Args: 

624 account: Account record. 

625 category: Email category, see ``core.Category``. 

626 args: Template arguments, the account record is added as ``account``. 

627 

628 Raises: 

629 Error: If the account has no email address. 

630 """ 

631 

632 

633 email = account.get(core.Columns.email) 

634 if not email: 

635 raise Error(f'account {self.get_uid(account)}: no email') 

636 

637 args = args or {} 

638 args['account'] = account 

639 

640 message = gws.plugin.email_helper.Message( 

641 subject=self.render_template(f'{category}.emailSubject', args), 

642 mailTo=email, 

643 text=self.render_template(f'{category}.emailBody', args, mime='text/plain'), 

644 html=self.render_template(f'{category}.emailBody', args, mime='text/html'), 

645 ) 

646 

647 email_helper = cast(gws.plugin.email_helper.Object, self.root.app.helper('email')) 

648 email_helper.send_mail(message) 

649 

650 def render_template(self, subject, args, mime: Optional[gws.MimeType] = None): 

651 """Render a template of this helper. 

652 

653 Args: 

654 subject: Template subject. 

655 args: Template arguments. 

656 mime: Template mime type. 

657 

658 Returns: 

659 The rendered content, or an empty string if there is no such template. 

660 """ 

661 

662 

663 tpl = self.root.app.templateMgr.find_template(subject, where=[self], mime=mime) 

664 if tpl: 

665 res = tpl.render(gws.TemplateRenderInput(args=args)) 

666 return res.content 

667 return ''