Coverage for app/auth/routes.py: 100%

1import logging 

2import os 

3import time 

4from datetime import UTC, datetime, timedelta 

5 

6import pw_hash as _pw 

7import pyotp 

8from extensions import _rate_limiting_disabled # pyright: ignore[reportMissingImports] 

9from extensions import cache as _cache 

10from extensions import limiter as _limiter 

11from flask import ( 

12 Blueprint, 

13 current_app, 

14 flash, 

15 redirect, 

16 render_template, 

17 request, 

18 session, 

19 url_for, 

20) 

21from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports] 

22from flask_babel import gettext as _ # pyright: ignore[reportMissingImports] 

23from markupsafe import Markup, escape 

24from models import ( 

25 OperatingModel, 

26 PasswordResetToken, 

27 Role, 

28 Tenant, 

29 TenantProfile, 

30 TenantUser, 

31 User, 

32 db, 

33) 

34from utils import login_required 

35 

36_log = logging.getLogger("openhangar.auth") 

37 

38 

39def _sl(value: object) -> str: 

40 """Sanitize a value for log output — strips CR/LF to prevent log injection (CWE-117).""" 

41 return str(value).replace("\r\n", "").replace("\n", "").replace("\r", "") 

42 

43 

44# Pre-computed Argon2id dummy hash used to equalise timing when no user record 

45# is found (prevents timing-based account enumeration — CWE-208). 

46_DUMMY_HASH: str = _pw.DUMMY_HASH 

47 

48auth_bp = Blueprint("auth", __name__) 

49 

50_COMPLEX_MODELS = frozenset({"shared_ownership", "flight_club", "flight_school"}) 

51 

52# ── Login brute-force protection ────────────────────────────────────────────── 

53# Two independent layers: 

54# 1. IP backoff — progressive delay applied to any IP accumulating failures 

55# 2. Account lock — 30-minute cache-based lock after 10 consecutive failures 

56# on the same e-mail address (auto-unlocks, no admin needed) 

57 

58_IP_FAIL_TTL = 900 # reset IP counter if silent for 15 min 

59_ACCT_FAIL_TTL = 1800 # reset account counter after 30 min 

60_ACCT_LOCK_MINUTES = 30 

61_ACCT_LOCK_TTL = _ACCT_LOCK_MINUTES * 60 

62_ACCT_LOCK_THRESHOLD = 10 

63 

64# Delay (seconds) applied before returning a failed-login response, keyed on 

65# the number of consecutive failures from the same IP address. 

66_IP_BACKOFF: dict[int, int] = {3: 2, 4: 10, 5: 30} 

67_IP_BACKOFF_MAX = 60 # applied for 6 or more failures 

68 

69 

70def _ip_backoff_delay(ip: str) -> int: 

71 count: int = _cache.get(f"login_fail_ip:{ip}") or 0 

72 if count < 3: 

73 return 0 

74 return _IP_BACKOFF.get(count, _IP_BACKOFF_MAX) 

75 

76 

77def _increment_ip_failures(ip: str) -> int: 

78 count: int = (_cache.get(f"login_fail_ip:{ip}") or 0) + 1 

79 _cache.set(f"login_fail_ip:{ip}", count, timeout=_IP_FAIL_TTL) 

80 return count 

81 

82 

83def _clear_ip_failures(ip: str) -> None: 

84 _cache.delete(f"login_fail_ip:{ip}") 

85 

86 

87def _check_account_locked(email: str) -> datetime | None: 

88 """Return the lock-expiry datetime if the account is locked, else None.""" 

89 raw = _cache.get(f"login_lock_acct:{email}") 

90 if not raw: 

91 return None 

92 locked_until = datetime.fromisoformat(raw) 

93 if datetime.now(UTC) < locked_until: 

94 return locked_until 

95 # Expired — clean up 

96 _cache.delete(f"login_lock_acct:{email}") 

97 _cache.delete(f"login_fail_acct:{email}") 

98 return None 

99 

100 

101def _increment_account_failures(email: str) -> int: 

102 count: int = (_cache.get(f"login_fail_acct:{email}") or 0) + 1 

103 _cache.set(f"login_fail_acct:{email}", count, timeout=_ACCT_FAIL_TTL) 

104 return count 

105 

106 

107def _lock_account(email: str, ip: str) -> None: 

108 locked_until = datetime.now(UTC) + timedelta(minutes=_ACCT_LOCK_MINUTES) 

109 _cache.set( 

110 f"login_lock_acct:{email}", 

111 locked_until.isoformat(), 

112 timeout=_ACCT_LOCK_TTL, 

113 ) 

114 _cache.delete(f"login_fail_acct:{email}") 

115 _log.warning( 

116 "[SECURITY] auth.login.account_locked email=%s ip=%s locked_until=%s", 

117 _sl(email), 

118 _sl(ip), 

119 _sl(locked_until.isoformat()), 

120 ) 

121 

122 

123def _clear_account_failures(email: str) -> None: 

124 _cache.delete(f"login_fail_acct:{email}") 

125 _cache.delete(f"login_lock_acct:{email}") 

126 

127 

128def _no_users() -> bool: 

129 return db.session.query(User).count() == 0 

130 

131 

132def _is_demo() -> bool: 

133 return os.environ.get("OPENHANGAR_ENV") == "demo" 

134 

135 

136# ── /login ──────────────────────────────────────────────────────────────────── 

137 

138 

139@auth_bp.route("/login", methods=["GET", "POST"]) 

140@_limiter.limit( 

141 lambda: current_app.config.get("LOGIN_RATE_LIMIT", "20 per minute"), 

142 methods=["POST"], 

143 exempt_when=_rate_limiting_disabled, 

144) 

145def login() -> ResponseReturnValue: 

146 if _no_users(): 

147 return redirect(url_for("auth.setup")) 

148 

149 if session.get("user_id"): 

150 return redirect(url_for("index")) 

151 

152 if request.method == "POST": 

153 return _login_post() 

154 

155 step = request.args.get("step", "credentials") 

156 # TOTP steps only accessible after credentials have been verified 

157 if step == "totp" and not session.get("login_pending_user_id"): 

158 return redirect(url_for("auth.login")) 

159 if step == "totp-enrol" and not session.get("totp_must_enrol"): 

160 return redirect(url_for("auth.login")) 

161 

162 totp_secret = session.get("enrol_totp_secret") 

163 totp_uri = session.get("enrol_totp_uri") 

164 return render_template( 

165 "auth/login.html", step=step, totp_secret=totp_secret, totp_uri=totp_uri 

166 ) 

167 

168 

169def _login_post() -> ResponseReturnValue: 

170 step = request.form.get("step") 

171 if step == "totp": 

172 return _login_totp() 

173 if step == "totp-enrol": 

174 return _login_totp_enrol() 

175 return _login_credentials() 

176 

177 

178def _login_credentials() -> ResponseReturnValue: 

179 email = request.form.get("email", "").strip().lower() 

180 password = request.form.get("password", "") 

181 ip = _sl(request.remote_addr or "") 

182 

183 # ── Account lockout check (before any verification) ────────────────────── 

184 locked_until = _check_account_locked(email) 

185 if locked_until: 

186 _log.warning( 

187 "[SECURITY] auth.login.account_blocked email=%s ip=%s locked_until=%s", 

188 _sl(email), 

189 ip, 

190 _sl(locked_until.isoformat()), 

191 ) 

192 flash( 

193 _( 

194 "Account temporarily locked due to too many failed attempts." 

195 " Try again in %(minutes)s minutes.", 

196 minutes=_ACCT_LOCK_MINUTES, 

197 ), 

198 "danger", 

199 ) 

200 return render_template("auth/login.html", step="credentials") 

201 

202 user = User.query.filter_by(email=email, is_active=True).first() 

203 password_hash = user.password_hash if user else _DUMMY_HASH 

204 # Always verify — never short-circuit on user is None. 

205 # Without this, a missing-user response is faster, enabling timing-based 

206 # account enumeration (CWE-208). pw_hash.verify handles both Argon2id and 

207 # legacy bcrypt hashes transparently. 

208 password_ok = _pw.verify(password, password_hash) 

209 

210 if not user or not password_ok: 

211 _log.warning( 

212 "[SECURITY] auth.credentials.failed email=%s ip=%s", 

213 _sl(email), 

214 ip, 

215 ) 

216 

217 # ── IP backoff ──────────────────────────────────────────────────────── 

218 ip_count = _increment_ip_failures(ip) 

219 delay = _ip_backoff_delay(ip) 

220 if delay: 

221 _log.warning( 

222 "[SECURITY] auth.login.backoff ip=%s failures=%s delay=%ss", 

223 ip, 

224 ip_count, 

225 delay, 

226 ) 

227 

228 # ── Account failure tracking ────────────────────────────────────────── 

229 acct_count = _increment_account_failures(email) 

230 if acct_count >= _ACCT_LOCK_THRESHOLD: 

231 _lock_account(email, ip) 

232 

233 if delay: 

234 time.sleep(delay) 

235 

236 flash(_("Invalid email or password."), "danger") 

237 return render_template("auth/login.html", step="credentials") 

238 

239 # Credentials verified — clear failure counters for this IP and account. 

240 _clear_ip_failures(ip) 

241 _clear_account_failures(email) 

242 

243 # Upgrade legacy bcrypt hashes to Argon2id transparently at login time. 

244 if _pw.needs_rehash(user.password_hash): 

245 user.password_hash = _pw.hash(password) 

246 db.session.commit() 

247 

248 # Block users whose only tenant(s) have been deactivated, unless they are 

249 # the instance admin (who must always be able to log in). 

250 if not user.is_instance_admin: 

251 active_tenant_ids = { 

252 tu.tenant_id for tu in user.tenants if tu.tenant and tu.tenant.is_active 

253 } 

254 if not active_tenant_ids: 

255 _log.warning( 

256 "[SECURITY] auth.credentials.deactivated email=%s ip=%s", 

257 _sl(email), 

258 _sl(request.remote_addr), 

259 ) 

260 flash( 

261 _("Your account has been deactivated. Contact the administrator."), 

262 "danger", 

263 ) 

264 return render_template("auth/login.html", step="credentials") 

265 

266 if user.totp_secret: 

267 session["login_pending_user_id"] = user.id 

268 return redirect(url_for("auth.login", step="totp")) 

269 

270 # If the tenant mandates TOTP and the user has none, redirect to enrolment. 

271 tu = TenantUser.query.filter_by(user_id=user.id).first() 

272 if tu and tu.tenant and tu.tenant.require_totp: 

273 totp_secret = pyotp.random_base32() 

274 totp_uri = pyotp.TOTP(totp_secret).provisioning_uri( 

275 name=user.email, issuer_name="OpenHangar" 

276 ) 

277 session["login_pending_user_id"] = user.id 

278 session["totp_must_enrol"] = True 

279 session["enrol_totp_secret"] = totp_secret 

280 session["enrol_totp_uri"] = totp_uri 

281 return redirect(url_for("auth.login", step="totp-enrol")) 

282 

283 session.clear() 

284 session["user_id"] = user.id 

285 session.permanent = True 

286 # _swr_fresh: tells the SW (sw.js) this / request must bypass whatever 

287 # it has cached (e.g. the landing page from before this login) and 

288 # re-cache the dashboard under the canonical / key. See pwa.js for the 

289 # matching history.replaceState() that scrubs the marker from the URL. 

290 return redirect(url_for("index", _swr_fresh=1)) 

291 

292 

293def _login_totp() -> ResponseReturnValue: 

294 pending_id = session.get("login_pending_user_id") 

295 if not pending_id: 

296 return redirect(url_for("auth.login")) 

297 

298 user = db.session.get(User, pending_id) 

299 if not user: 

300 session.pop("login_pending_user_id", None) 

301 return redirect(url_for("auth.login")) 

302 

303 totp_code = request.form.get("totp_code", "").strip() 

304 

305 _totp_cache_key = f"totp_used:{user.id}:{totp_code}" 

306 if _cache.get(_totp_cache_key): 

307 _log.warning( 

308 "[SECURITY] auth.totp.replay user_id=%s ip=%s", 

309 _sl(pending_id), 

310 _sl(request.remote_addr), 

311 ) 

312 flash(_("Invalid authenticator code."), "danger") 

313 return render_template("auth/login.html", step="totp") 

314 

315 if not pyotp.TOTP(str(user.totp_secret)).verify(totp_code, valid_window=1): 

316 _log.warning( 

317 "[SECURITY] auth.totp.failed user_id=%s ip=%s", 

318 _sl(pending_id), 

319 _sl(request.remote_addr), 

320 ) 

321 flash(_("Invalid authenticator code."), "danger") 

322 return render_template("auth/login.html", step="totp") 

323 

324 _cache.set(_totp_cache_key, True, timeout=90) 

325 

326 session.clear() 

327 session["user_id"] = user.id 

328 session.permanent = True 

329 # _swr_fresh: tells the SW (sw.js) this / request must bypass whatever 

330 # it has cached (e.g. the landing page from before this login) and 

331 # re-cache the dashboard under the canonical / key. See pwa.js for the 

332 # matching history.replaceState() that scrubs the marker from the URL. 

333 return redirect(url_for("index", _swr_fresh=1)) 

334 

335 

336def _login_totp_enrol() -> ResponseReturnValue: 

337 """Mandatory TOTP enrolment during login when tenant.require_totp is True.""" 

338 pending_id = session.get("login_pending_user_id") 

339 totp_secret = session.get("enrol_totp_secret") 

340 totp_uri = session.get("enrol_totp_uri") 

341 if not pending_id or not totp_secret: 

342 return redirect(url_for("auth.login")) 

343 

344 user = db.session.get(User, pending_id) 

345 if not user: 

346 session.clear() 

347 return redirect(url_for("auth.login")) 

348 

349 totp_code = request.form.get("totp_code", "").strip() 

350 if not pyotp.TOTP(totp_secret).verify(totp_code, valid_window=1): 

351 flash(_("Invalid code — please try again."), "danger") 

352 return render_template( 

353 "auth/login.html", 

354 step="totp-enrol", 

355 totp_secret=totp_secret, 

356 totp_uri=totp_uri, 

357 ) 

358 

359 user.totp_secret = totp_secret 

360 db.session.commit() 

361 _log.warning( 

362 "[SECURITY] auth.totp.enrolment_forced user_id=%s ip=%s", 

363 _sl(str(user.id)), 

364 _sl(request.remote_addr), 

365 ) 

366 

367 session.clear() 

368 session["user_id"] = user.id 

369 session.permanent = True 

370 flash(_("Two-factor authentication is now active on your account."), "success") 

371 # See the _swr_fresh comment on the other two login-success redirects above. 

372 return redirect(url_for("index", _swr_fresh=1)) 

373 

374 

375# ── /logout ─────────────────────────────────────────────────────────────────── 

376 

377 

378@auth_bp.route("/logout") 

379def logout() -> ResponseReturnValue: 

380 slot_id = session.get("demo_slot_id") 

381 session.clear() 

382 if slot_id is not None: 

383 # Preserve the slot so the visitor can re-enter the same sandbox 

384 session["demo_slot_id"] = slot_id 

385 return redirect(url_for("index")) 

386 

387 

388# ── /setup ──────────────────────────────────────────────────────────────────── 

389 

390_WIZARD_STEPS = [ 

391 "account", 

392 "totp", 

393 "operating_model", 

394 "aircraft_count", 

395 "org_name", 

396 "co_owners", 

397 "summary", 

398] 

399 

400_OPERATING_MODELS = { 

401 OperatingModel.SOLE_PILOT, 

402 OperatingModel.SOLE_OPERATOR, 

403 OperatingModel.SHARED_OWNERSHIP, 

404 OperatingModel.FLIGHT_CLUB, 

405 OperatingModel.FLIGHT_SCHOOL, 

406} 

407 

408 

409def _wizard_phase(step: str) -> int: 

410 """Map a wizard step to a 1-based display phase (for the progress indicator).""" 

411 if step in ("account", "totp"): 

412 return 1 

413 if step == "operating_model": 

414 return 2 

415 if step in ("aircraft_count", "org_name", "co_owners"): 

416 return 3 

417 return 4 # summary 

418 

419 

420def _next_step(current: str) -> str: 

421 """Compute the next wizard step based on current step and session choices.""" 

422 operating_model = session.get("setup_operating_model", "") 

423 

424 if current == "account": # pragma: no cover 

425 return "totp" 

426 if current == "totp": # pragma: no cover 

427 return "operating_model" 

428 if current == "operating_model": # pragma: no cover 

429 return "summary" if operating_model == "sole_pilot" else "aircraft_count" 

430 if current == "aircraft_count": 

431 if operating_model in ("flight_club", "flight_school"): 

432 return "org_name" 

433 if operating_model == "shared_ownership": 

434 return "co_owners" 

435 return "summary" 

436 if current in ("org_name", "co_owners"): # pragma: no cover 

437 return "summary" 

438 return "summary" # pragma: no cover 

439 

440 

441@auth_bp.route("/setup", methods=["GET", "POST"]) 

442@_limiter.limit("10 per minute", methods=["POST"], exempt_when=_rate_limiting_disabled) 

443def setup() -> ResponseReturnValue: 

444 if _is_demo(): 

445 flash(_("Account creation is disabled in demo mode."), "warning") 

446 return redirect(url_for("index")) 

447 

448 if not _no_users(): 

449 return redirect(url_for("config.index")) 

450 

451 # Determine current step from form data (POST) or query string (GET) 

452 step = request.form.get("step") or request.args.get("step", "account") 

453 if step not in _WIZARD_STEPS: 

454 return redirect(url_for("auth.setup")) 

455 

456 if request.method == "POST": 

457 if step == "account": 

458 return _setup_account() 

459 if step == "totp": 

460 return _setup_totp() 

461 if step == "operating_model": 

462 return _setup_operating_model() 

463 if step == "aircraft_count": 

464 return _setup_aircraft_count() 

465 if step == "org_name": 

466 return _setup_org_name() 

467 if step == "co_owners": 

468 return _setup_co_owners() 

469 if step == "summary": 

470 return _setup_finish() 

471 

472 # GET handlers — validate session state before rendering each step 

473 phase = _wizard_phase(step) 

474 

475 if step == "totp": 

476 if not session.get("setup_totp_secret"): 

477 return redirect(url_for("auth.setup")) 

478 return render_template( 

479 "auth/setup.html", 

480 step="totp", 

481 phase=phase, 

482 show_review=False, 

483 totp_secret=session["setup_totp_secret"], 

484 provisioning_uri=session["setup_provisioning_uri"], 

485 ) 

486 

487 if step == "operating_model": 

488 if not session.get("setup_totp_done"): 

489 return redirect(url_for("auth.setup")) 

490 return render_template( 

491 "auth/setup.html", step="operating_model", phase=phase, show_review=False 

492 ) 

493 

494 if step == "aircraft_count": 

495 if not session.get("setup_operating_model"): 

496 return redirect(url_for("auth.setup", step="operating_model")) 

497 return render_template( 

498 "auth/setup.html", 

499 step="aircraft_count", 

500 phase=phase, 

501 show_review=session.get("setup_operating_model") in _COMPLEX_MODELS, 

502 operating_model=session.get("setup_operating_model"), 

503 ) 

504 

505 if step == "org_name": 

506 model = session.get("setup_operating_model", "") 

507 if model not in ("flight_club", "flight_school"): 

508 return redirect(url_for("auth.setup", step="summary")) 

509 return render_template( 

510 "auth/setup.html", 

511 step="org_name", 

512 phase=phase, 

513 show_review=True, 

514 operating_model=model, 

515 ) 

516 

517 if step == "co_owners": 

518 if session.get("setup_operating_model") != "shared_ownership": 

519 return redirect(url_for("auth.setup", step="summary")) 

520 return render_template( 

521 "auth/setup.html", step="co_owners", phase=phase, show_review=True 

522 ) 

523 

524 if step == "summary": 

525 if not session.get("setup_operating_model"): 

526 return redirect(url_for("auth.setup", step="operating_model")) 

527 if session.get("setup_operating_model") in ("sole_pilot", "sole_operator"): 

528 return redirect(url_for("auth.setup", step="operating_model")) 

529 return render_template( 

530 "auth/setup.html", 

531 step="summary", 

532 phase=phase, 

533 show_review=True, 

534 operating_model=session.get("setup_operating_model"), 

535 aircraft_count=session.get("setup_aircraft_count"), 

536 allows_rental=session.get("setup_allows_rental", False), 

537 org_name=session.get("setup_org_name", ""), 

538 co_owners=session.get("setup_co_owners", []), 

539 setup_name=session.get("setup_name", ""), 

540 setup_email=session.get("setup_email", ""), 

541 ) 

542 

543 return render_template( 

544 "auth/setup.html", step="account", phase=1, show_review=False 

545 ) 

546 

547 

548def _setup_account() -> ResponseReturnValue: 

549 email = request.form.get("email", "").strip().lower() 

550 password = request.form.get("password", "") 

551 name = request.form.get("name", "").strip() 

552 

553 errors = [] 

554 if not email or "@" not in email: 

555 errors.append(_("A valid email address is required.")) 

556 if len(password) < 12: 

557 errors.append(_("Password must be at least 12 characters.")) 

558 

559 if errors: 

560 for msg in errors: 

561 flash(msg, "danger") 

562 return render_template( 

563 "auth/setup.html", step="account", phase=1, show_review=False 

564 ) 

565 

566 totp_secret = pyotp.random_base32() 

567 provisioning_uri = pyotp.TOTP(totp_secret).provisioning_uri( 

568 name=email, issuer_name="OpenHangar" 

569 ) 

570 

571 session["setup_email"] = email 

572 session["setup_name"] = name or None 

573 session["setup_password_hash"] = _pw.hash(password) 

574 session["setup_totp_secret"] = totp_secret 

575 session["setup_provisioning_uri"] = provisioning_uri 

576 

577 return redirect(url_for("auth.setup", step="totp")) 

578 

579 

580def _setup_totp() -> ResponseReturnValue: 

581 email = session.get("setup_email") 

582 password_hash = session.get("setup_password_hash") 

583 totp_secret = session.get("setup_totp_secret") 

584 provisioning_uri = session.get("setup_provisioning_uri") 

585 

586 if not all([email, password_hash, totp_secret]): 

587 flash(_("Session expired. Please start over."), "danger") 

588 return redirect(url_for("auth.setup")) 

589 

590 # "Skip" path — user will not have TOTP 

591 if request.form.get("action") == "skip": 

592 session["setup_totp_to_save"] = None 

593 else: 

594 totp_code = request.form.get("totp_code", "").strip() 

595 if not pyotp.TOTP(str(totp_secret)).verify(totp_code, valid_window=1): 

596 flash(_("Invalid code. Please try again."), "danger") 

597 return render_template( 

598 "auth/setup.html", 

599 step="totp", 

600 phase=1, 

601 show_review=False, 

602 totp_secret=totp_secret, 

603 provisioning_uri=provisioning_uri, 

604 ) 

605 session["setup_totp_to_save"] = totp_secret 

606 

607 session["setup_totp_done"] = True 

608 return redirect(url_for("auth.setup", step="operating_model")) 

609 

610 

611def _setup_operating_model() -> ResponseReturnValue: 

612 if not session.get("setup_totp_done"): 

613 return redirect(url_for("auth.setup")) 

614 

615 model = request.form.get("operating_model", "") 

616 valid = {m.value for m in _OPERATING_MODELS} 

617 if model not in valid: 

618 flash(_("Please select an option."), "danger") 

619 return render_template( 

620 "auth/setup.html", step="operating_model", phase=2, show_review=False 

621 ) 

622 

623 session["setup_operating_model"] = model 

624 if model == "sole_pilot": 

625 return _setup_finish() 

626 return redirect(url_for("auth.setup", step="aircraft_count")) 

627 

628 

629def _setup_aircraft_count() -> ResponseReturnValue: 

630 if not session.get("setup_operating_model"): 

631 return redirect(url_for("auth.setup", step="operating_model")) 

632 

633 count_str = request.form.get("aircraft_count", "").strip() 

634 try: 

635 count = int(count_str) 

636 if count < 0: 

637 raise ValueError 

638 except (ValueError, TypeError): 

639 flash(_("Please enter a valid number of aircraft (0 or more)."), "danger") 

640 return render_template( 

641 "auth/setup.html", 

642 step="aircraft_count", 

643 phase=3, 

644 show_review=session.get("setup_operating_model") in _COMPLEX_MODELS, 

645 operating_model=session.get("setup_operating_model"), 

646 ) 

647 

648 allows_rental = "allows_rental" in request.form 

649 session["setup_aircraft_count"] = count 

650 session["setup_allows_rental"] = allows_rental 

651 

652 next_step = _next_step("aircraft_count") 

653 if next_step == "summary": 

654 return _setup_finish() 

655 return redirect(url_for("auth.setup", step=next_step)) 

656 

657 

658def _setup_org_name() -> ResponseReturnValue: 

659 model = session.get("setup_operating_model", "") 

660 if model not in ("flight_club", "flight_school"): 

661 return redirect(url_for("auth.setup", step="summary")) 

662 

663 org_name = request.form.get("org_name", "").strip() 

664 if not org_name: 

665 flash(_("Please enter a name."), "danger") 

666 return render_template( 

667 "auth/setup.html", 

668 step="org_name", 

669 phase=3, 

670 show_review=True, 

671 operating_model=model, 

672 ) 

673 

674 session["setup_org_name"] = org_name 

675 return redirect(url_for("auth.setup", step="summary")) 

676 

677 

678def _setup_co_owners() -> ResponseReturnValue: 

679 if session.get("setup_operating_model") != "shared_ownership": 

680 return redirect(url_for("auth.setup", step="summary")) 

681 

682 names = request.form.getlist("co_owner_name") 

683 emails = request.form.getlist("co_owner_email") 

684 roles = request.form.getlist("co_owner_role") 

685 

686 co_owners = [] 

687 for name, email, role in zip(names, emails, roles): 

688 name = name.strip() 

689 email = email.strip().lower() 

690 role = role if role in ("owner", "admin") else "owner" 

691 if name or email: 

692 co_owners.append( 

693 {"name": name or None, "email": email or None, "role": role} 

694 ) 

695 

696 session["setup_co_owners"] = co_owners 

697 return redirect(url_for("auth.setup", step="summary")) 

698 

699 

700def _setup_finish() -> ResponseReturnValue: 

701 required = ["setup_email", "setup_password_hash", "setup_operating_model"] 

702 if not all(session.get(k) for k in required) or not session.get("setup_totp_done"): 

703 flash(_("Session expired. Please start over."), "danger") 

704 return redirect(url_for("auth.setup")) 

705 

706 from models import UserInvitation 

707 

708 operating_model_raw = session.get("setup_operating_model", "") 

709 aircraft_count = session.get("setup_aircraft_count") 

710 allows_rental = bool(session.get("setup_allows_rental", False)) 

711 org_name = session.get("setup_org_name", "") 

712 co_owners = session.get("setup_co_owners", []) 

713 

714 # Choose tenant name based on operating model 

715 tenant_name = "My Hangar" 

716 if operating_model_raw in ("flight_club", "flight_school") and org_name: 

717 tenant_name = org_name 

718 

719 tenant = Tenant(name=tenant_name) 

720 db.session.add(tenant) 

721 db.session.flush() 

722 

723 # Auto-generate the Hangar ID from the tenant name so canonical document 

724 # paths work immediately, without requiring a trip to Settings first. 

725 from documents.routes import ( 

726 _ensure_tenant_slug, # pyright: ignore[reportMissingImports] 

727 ) 

728 

729 _ensure_tenant_slug(tenant) 

730 

731 user = User( 

732 email=session["setup_email"], 

733 password_hash=session["setup_password_hash"], 

734 totp_secret=session.get("setup_totp_to_save"), 

735 name=session.get("setup_name"), 

736 is_active=True, 

737 is_instance_admin=True, 

738 ) 

739 db.session.add(user) 

740 db.session.flush() 

741 

742 db.session.add(TenantUser(user_id=user.id, tenant_id=tenant.id, role=Role.ADMIN)) 

743 

744 # Determine profile values from wizard 

745 try: 

746 op_model: OperatingModel | None = OperatingModel(operating_model_raw) 

747 except ValueError: 

748 op_model = None 

749 planned_count: int | None = ( 

750 0 if operating_model_raw == "sole_pilot" else aircraft_count 

751 ) 

752 

753 club_name = org_name if operating_model_raw == "flight_club" else None 

754 school_name = org_name if operating_model_raw == "flight_school" else None 

755 

756 profile = TenantProfile( 

757 tenant_id=tenant.id, 

758 operating_model=op_model, 

759 planned_aircraft_count=planned_count, 

760 allows_rental=allows_rental, 

761 club_name=club_name, 

762 school_name=school_name, 

763 setup_complete=True, 

764 ) 

765 db.session.add(profile) 

766 

767 # Create co-owner invitations (shared_ownership path) 

768 for co in co_owners: 

769 inv_role = Role.ADMIN if co.get("role") == "admin" else Role.OWNER 

770 inv = UserInvitation( 

771 tenant_id=tenant.id, 

772 invited_by_user_id=user.id, 

773 email=co.get("email") or None, 

774 display_name=co.get("name") or None, 

775 role=inv_role, 

776 expires_at=datetime.now(UTC) + timedelta(days=7), 

777 ) 

778 db.session.add(inv) 

779 

780 db.session.commit() 

781 

782 aircraft_url = url_for("aircraft.new_aircraft") 

783 flight_url = url_for("flights.log_flight") 

784 is_pilot_ctx = operating_model_raw in ( 

785 "sole_pilot", 

786 "sole_operator", 

787 "shared_ownership", 

788 ) 

789 is_operator_ctx = operating_model_raw != "sole_pilot" 

790 welcome = escape(_("Setup complete. Welcome to OpenHangar!")) 

791 aircraft_label = _("Add your first aircraft")