Coverage for app/config/routes.py: 100%
864 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 15:29 +0000
1"""
2Configuration blueprint — backup management, email settings, and future config sections.
3"""
5import contextlib
6import hashlib
7import io
8import json
9import logging
10import os
11import subprocess # nosec B404
12import urllib.error
13import urllib.request
14import zipfile
15from datetime import UTC, datetime
16from typing import Any
18from flask import (
19 Blueprint,
20 abort,
21 current_app,
22 flash,
23 jsonify,
24 redirect,
25 render_template,
26 request,
27 session,
28 url_for,
29) # pyright: ignore[reportMissingImports]
30from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
31from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
32from flask_babel import ngettext
33from init import _env_or_file # pyright: ignore[reportMissingImports]
34from models import AppSetting, BackupRecord, db # pyright: ignore[reportMissingImports]
35from utils import ( # pyright: ignore[reportMissingImports]
36 login_required,
37 require_instance_admin,
38 to_libpq_url,
39)
41config_bp = Blueprint("config", __name__, url_prefix="/config")
42log = logging.getLogger(__name__)
45# ── helpers ───────────────────────────────────────────────────────────────────
48def _derive_key(passphrase: str) -> bytes:
49 """Derive a 32-byte AES key from a passphrase using HKDF-SHA256."""
50 from cryptography.hazmat.primitives import (
51 hashes, # pyright: ignore[reportMissingImports]
52 )
53 from cryptography.hazmat.primitives.kdf.hkdf import (
54 HKDF, # pyright: ignore[reportMissingImports]
55 )
57 return HKDF(
58 algorithm=hashes.SHA256(),
59 length=32,
60 salt=b"openhangar-backup-kdf-salt-v1",
61 info=b"openhangar-backup-v1",
62 ).derive(passphrase.encode())
65def _encrypt_bytes(plaintext: bytes, key: bytes) -> bytes:
66 """Encrypt *plaintext* with AES-256-GCM, prepending the 12-byte nonce."""
67 import os as _os
69 from cryptography.hazmat.primitives.ciphers.aead import (
70 AESGCM, # pyright: ignore[reportMissingImports]
71 )
73 nonce = _os.urandom(12)
74 ct = AESGCM(key).encrypt(nonce, plaintext, None)
75 return nonce + ct
78def _get_alembic_head() -> str | None:
79 """Return current Alembic revision from the DB, or None if unavailable."""
80 try:
81 from sqlalchemy import text # pyright: ignore[reportMissingImports]
83 return db.session.execute(
84 text("SELECT version_num FROM alembic_version LIMIT 1")
85 ).scalar()
86 except Exception: # noqa: BLE001 -- optional diagnostics probe, any driver/SQL error means "unknown"
87 return None
90def _parse_gatus_env() -> tuple[str, str, str | None] | None:
91 """Return (base_url, endpoint_key, auth_header_or_None) from env vars, or None if not configured."""
92 endpoint_url = os.environ.get("OPENHANGAR_GATUS_ENDPOINT_URL", "").rstrip("/")
93 if not endpoint_url or "/endpoints/" not in endpoint_url:
94 return None
95 base_url, _, endpoint_key = endpoint_url.rpartition("/endpoints/")
96 if not base_url or not endpoint_key:
97 return None
98 auth_header = _env_or_file("GATUS_AUTH_HEADER") or None
99 return base_url, endpoint_key, auth_header
102def run_backup() -> BackupRecord:
103 """
104 Produce an encrypted ZIP backup of the PostgreSQL database and uploaded
105 documents.
107 The ZIP contains:
108 - ``openhangar.sql`` — full pg_dump output
109 - ``uploads/<filename>`` — every file from the uploads folder
111 The ZIP is then AES-256-GCM encrypted and written to the backup folder.
112 A ``BackupRecord`` row is committed and returned.
114 Raises ``RuntimeError`` on failure; the record is still committed with
115 ``status='failed'`` so operators can see the attempt.
116 """
117 from flask import current_app # pyright: ignore[reportMissingImports]
119 backup_folder = current_app.config.get("BACKUP_FOLDER", "/data/backups")
120 upload_folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
121 encryption_key_raw = _env_or_file("BACKUP_ENCRYPTION_KEY")
122 database_url = current_app.config.get("SQLALCHEMY_DATABASE_URI", "")
124 os.makedirs(backup_folder, exist_ok=True)
126 ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
127 app_version = os.environ.get("OPENHANGAR_VERSION", "development")
128 filename = f"openhangar_backup_{ts}_{app_version}.zip.enc"
129 path = os.path.join(backup_folder, filename)
130 alembic_head = _get_alembic_head()
131 metadata = {
132 "app_version": app_version,
133 "alembic_head": alembic_head,
134 "created_at": datetime.now(UTC).isoformat(),
135 }
137 record = BackupRecord(
138 filename=filename,
139 path=path,
140 status="failed",
141 app_version=app_version,
142 alembic_head=alembic_head,
143 )
144 db.session.add(record)
145 db.session.flush() # get an id without committing
147 try:
148 sql_bytes = _pg_dump(database_url)
150 buf = io.BytesIO()
151 with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
152 zf.writestr("openhangar.sql", sql_bytes)
153 zf.writestr("metadata.json", json.dumps(metadata, indent=2))
154 _add_uploads_to_zip(zf, upload_folder)
155 zip_bytes = buf.getvalue()
157 if encryption_key_raw:
158 key = _derive_key(encryption_key_raw)
159 payload = _encrypt_bytes(zip_bytes, key)
160 else:
161 payload = zip_bytes
162 log.warning(
163 "OPENHANGAR_BACKUP_ENCRYPTION_KEY not set — backup is unencrypted"
164 )
166 with open(path, "wb") as fh:
167 fh.write(payload)
169 meta_path = path.replace(".zip.enc", ".meta")
170 with open(meta_path, "w") as fh:
171 json.dump(metadata, fh, indent=2)
173 sha256 = hashlib.sha256(payload).hexdigest()
174 record.size_bytes = len(payload)
175 record.sha256 = sha256
176 record.status = "ok"
177 except Exception as exc:
178 log.error("Backup failed: %s", exc)
179 db.session.commit()
180 raise RuntimeError(str(exc)) from exc
182 db.session.commit()
183 return record
186def _add_uploads_to_zip(zf: zipfile.ZipFile, upload_folder: str) -> None:
187 """Add every file in *upload_folder* into the ZIP under ``uploads/``, preserving subdirs."""
188 if not os.path.isdir(upload_folder):
189 return
190 for dirpath, _dirs, filenames in os.walk(upload_folder):
191 for fname in filenames:
192 full_path = os.path.join(dirpath, fname)
193 rel_path = os.path.relpath(full_path, upload_folder)
194 zf.write(full_path, arcname=f"uploads/{rel_path}")
197def _pg_dump(database_url: str) -> bytes:
198 """Run pg_dump against *database_url* and return the SQL as bytes."""
199 env = os.environ.copy()
200 if database_url.startswith("postgresql"):
201 libpq_url = to_libpq_url(database_url)
202 env["DATABASE_URL"] = libpq_url
203 # pg_dump reads PGPASSWORD / connection string
204 cmd = [
205 "pg_dump",
206 "--no-password",
207 "--no-owner", # omit ALTER … OWNER TO: role names are environment-specific
208 "--no-acl", # omit GRANT/REVOKE: privileges are managed by the app, not the DB
209 libpq_url,
210 ]
211 else:
212 raise RuntimeError(f"Unsupported database URL scheme: {database_url!r}")
214 result = subprocess.run( # nosec B603 # fixed list, no shell, DB URL from server config
215 cmd,
216 capture_output=True,
217 env=env,
218 timeout=120,
219 check=False, # returncode checked explicitly below
220 )
221 if result.returncode != 0:
222 raise RuntimeError(result.stderr.decode(errors="replace"))
223 return result.stdout
226# ── views ─────────────────────────────────────────────────────────────────────
229@config_bp.before_request
230def _block_in_demo() -> None:
231 if os.environ.get("OPENHANGAR_ENV") == "demo":
232 abort(403)
233 if session.get("user_id"):
234 # All logged-in users may manage their own notification preferences
235 if request.endpoint == "config.notification_preferences":
236 return
237 from models import Role, User # pyright: ignore[reportMissingImports]
238 from utils import current_user_role # pyright: ignore[reportMissingImports]
240 user = db.session.get(User, session["user_id"])
241 # Instance admins always pass — they may not have a tenant role
242 if user and user.is_instance_admin:
243 return
244 if current_user_role() not in (Role.ADMIN, Role.OWNER):
245 abort(403)
248@config_bp.route("/")
249def index() -> ResponseReturnValue:
250 if not session.get("user_id"):
251 return redirect(url_for("auth.login"))
252 from services.email_service import ( # pyright: ignore[reportMissingImports]
253 get_email_health,
254 get_smtp_status,
255 )
257 _BACKUP_DISPLAY_LIMIT = 10
258 total_backups = BackupRecord.query.count()
259 records = (
260 BackupRecord.query.order_by(BackupRecord.created_at.desc())
261 .limit(_BACKUP_DISPLAY_LIMIT)
262 .all()
263 )
264 backup_extra = max(0, total_backups - _BACKUP_DISPLAY_LIMIT)
266 # Built-in backup scheduling status. Parse errors are swallowed here —
267 # startup validation already reports them; the page shows "not scheduled".
268 from datetime import timedelta as _timedelta
270 from services.backup_scheduler import ( # pyright: ignore[reportMissingImports]
271 RETENTION_GFS,
272 parse_auto_upgrade_enabled,
273 parse_backup_keep,
274 parse_backup_keep_days,
275 parse_backup_keep_months,
276 parse_backup_keep_weeks,
277 parse_backup_retention,
278 parse_backup_time,
279 )
281 try:
282 _schedule = parse_backup_time()
283 except ValueError:
284 _schedule = None
285 try:
286 auto_upgrade_enabled = parse_auto_upgrade_enabled()
287 except ValueError:
288 auto_upgrade_enabled = False
289 backup_schedule_str = (
290 f"{_schedule[0]:02d}:{_schedule[1]:02d}" if _schedule else None
291 )
292 try:
293 backup_keep = parse_backup_keep()
294 except ValueError:
295 backup_keep = None
296 backup_gfs = None
297 try:
298 if parse_backup_retention() == RETENTION_GFS:
299 backup_gfs = {
300 "days": parse_backup_keep_days(),
301 "weeks": parse_backup_keep_weeks(),
302 "months": parse_backup_keep_months(),
303 }
304 except ValueError:
305 backup_gfs = None
306 _last_ok = (
307 BackupRecord.query.filter_by(status="ok")
308 .order_by(BackupRecord.created_at.desc())
309 .first()
310 )
311 last_backup_at = _last_ok.created_at if _last_ok else None
312 if last_backup_at is not None and last_backup_at.tzinfo is None:
313 # SQLite returns naive datetimes; values are stored in UTC.
314 last_backup_at = last_backup_at.replace(tzinfo=UTC)
315 backup_stale = backup_schedule_str is not None and (
316 last_backup_at is None
317 or datetime.now(UTC) - last_backup_at > _timedelta(days=2)
318 )
320 from models import ( # pyright: ignore[reportMissingImports]
321 Role,
322 TenantUser,
323 User,
324 UserInvitation,
325 )
326 from sqlalchemy import func # pyright: ignore[reportMissingImports]
327 from users.routes import ROLE_LABELS # pyright: ignore[reportMissingImports]
329 tu_self = TenantUser.query.filter_by(user_id=session["user_id"]).first()
330 tid = tu_self.tenant_id if tu_self else None
331 user_counts = []
332 open_invitations = 0
333 if tid:
334 results = (
335 db.session.query(TenantUser.role, func.count(TenantUser.user_id))
336 .join(User, TenantUser.user_id == User.id)
337 .filter(TenantUser.tenant_id == tid, User.is_active.is_(True))
338 .group_by(TenantUser.role)
339 .all()
340 )
341 counts_by_role = dict(results)
342 user_counts = [
343 (ROLE_LABELS[r], counts_by_role[r])
344 for r in Role
345 if counts_by_role.get(r, 0) > 0
346 ]
347 open_invitations = (
348 UserInvitation.query.filter_by(tenant_id=tid)
349 .filter(UserInvitation.accepted_at.is_(None))
350 .count()
351 )
352 current_version = os.environ.get("OPENHANGAR_VERSION", "development")
353 latest_setting = db.session.get(AppSetting, "latest_version")
354 latest_version = latest_setting.value if latest_setting else None
355 from utils import check_update_available # pyright: ignore[reportMissingImports]
357 update_available = check_update_available()
358 versions_behind: int | None = None
359 try:
360 import json as _json
362 _all_v_setting = db.session.get(AppSetting, "all_versions")
363 if _all_v_setting and current_version != "development":
364 _all_versions = _json.loads(_all_v_setting.value)
365 if isinstance(_all_versions, list) and current_version in _all_versions:
366 _idx = _all_versions.index(current_version)
367 if _idx > 0:
368 versions_behind = _idx
369 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
370 log.debug("Could not compute versions-behind count: %s", exc)
371 db_size: str | None = None
372 try:
373 from sqlalchemy import text as _text # pyright: ignore[reportMissingImports]
375 _res = db.session.execute(
376 _text("SELECT pg_size_pretty(pg_database_size(current_database()))")
377 ).scalar()
378 db_size = str(_res) if _res is not None else None
379 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
380 log.debug("Could not retrieve DB size: %s", exc)
381 upload_size_bytes: int | None = None
382 try:
383 _upload_folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
384 if os.path.isdir(_upload_folder):
385 upload_size_bytes = sum(
386 os.path.getsize(os.path.join(dp, f))
387 for dp, _dirs, files in os.walk(_upload_folder)
388 for f in files
389 )
390 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
391 log.debug("Could not retrieve upload folder size: %s", exc)
392 backup_total_size_bytes: int | None = None
393 backup_file_count: int = 0
394 try:
395 _backup_folder = current_app.config.get("BACKUP_FOLDER", "/data/backups")
396 if os.path.isdir(_backup_folder):
397 _backup_sizes = [
398 os.path.getsize(os.path.join(dp, f))
399 for dp, _dirs, files in os.walk(_backup_folder)
400 for f in files
401 ]
402 backup_total_size_bytes = sum(_backup_sizes)
403 backup_file_count = len(_backup_sizes)
404 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
405 log.debug("Could not retrieve backup folder size: %s", exc)
406 from models import Tenant, User # pyright: ignore[reportMissingImports]
408 current_user = db.session.get(User, session["user_id"])
409 tenant_count = Tenant.query.count()
410 _tenant = db.session.get(Tenant, tid) if tid else None
411 upgrade_dir = os.environ.get("OPENHANGAR_UPGRADE_DIR", "")
412 upgrade_dir_enabled = bool(upgrade_dir)
413 upgrade_active = False
414 if upgrade_dir:
415 # Catches an unattended auto-upgrade's completion marker even if no
416 # browser was polling /upgrade-status when it finished — see
417 # _record_last_upgrade's docstring.
418 _record_last_upgrade(upgrade_dir)
419 upgrade_active = os.path.exists(
420 os.path.join(upgrade_dir, "trigger")
421 ) or os.path.exists(os.path.join(upgrade_dir, "trigger.running"))
422 last_upgrade_setting = db.session.get(AppSetting, "last_upgrade_at")
423 last_upgrade_at = last_upgrade_setting.value if last_upgrade_setting else None
424 app_debug = current_app.debug
425 sw_forced_on = os.environ.get("OPENHANGAR_SW_ENABLED", "").lower() in (
426 "1",
427 "true",
428 "yes",
429 )
430 sw_server_enabled = not app_debug or sw_forced_on
431 # "demo" is excluded here — _block_in_demo() aborts the whole blueprint
432 # with a 403 before this view runs when OPENHANGAR_ENV=demo.
433 _env_labels = {
434 "production": _("Production"),
435 "development": _("Development"),
436 "test": _("Test"),
437 }
438 _flask_env = os.environ.get("OPENHANGAR_ENV", "production")
439 env_label = _env_labels.get(_flask_env, _flask_env)
440 return render_template(
441 "config/settings.html",
442 records=records,
443 backup_extra=backup_extra,
444 backup_encryption_key_set=bool(_env_or_file("BACKUP_ENCRYPTION_KEY")),
445 backup_folder=current_app.config.get("BACKUP_FOLDER", "/data/backups"),
446 backup_schedule_str=backup_schedule_str,
447 backup_keep=backup_keep,
448 backup_gfs=backup_gfs,
449 last_backup_at=last_backup_at,
450 backup_stale=backup_stale,
451 smtp_status=get_smtp_status(),
452 email_health=get_email_health(),
453 user_counts=user_counts,
454 open_invitations=open_invitations,
455 current_version=current_version,
456 latest_version=latest_version,
457 update_available=update_available,
458 versions_behind=versions_behind,
459 auto_upgrade_enabled=auto_upgrade_enabled,
460 last_upgrade_at=last_upgrade_at,
461 db_size=db_size,
462 upload_size_bytes=upload_size_bytes,
463 backup_total_size_bytes=backup_total_size_bytes,
464 backup_file_count=backup_file_count,
465 current_user=current_user,
466 tenant_count=tenant_count,
467 tenant=_tenant,
468 openaip_api_key=(
469 db.session.get(AppSetting, "openaip_api_key")
470 or type("_", (), {"value": None})()
471 ).value,
472 gatus_configured=_parse_gatus_env() is not None,
473 upgrade_dir_enabled=upgrade_dir_enabled,
474 upgrade_active=upgrade_active,
475 app_debug=app_debug,
476 sw_server_enabled=sw_server_enabled,
477 env_label=env_label,
478 )
481@config_bp.route("/tenant-slug", methods=["POST"])
482@login_required
483def update_tenant_slug() -> ResponseReturnValue:
484 import re as _re
485 import shutil
487 from models import ( # pyright: ignore[reportMissingImports]
488 AircraftPhoto,
489 Document,
490 PendingReconcile,
491 Tenant,
492 TenantUser,
493 )
495 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
496 if not tu:
497 abort(403) # pragma: no cover
498 tenant = db.session.get(Tenant, tu.tenant_id)
499 if not tenant:
500 abort(403) # pragma: no cover
502 raw = request.form.get("slug", "").strip().lower()
503 if not raw:
504 flash(_("Hangar ID cannot be empty."), "danger")
505 return redirect(url_for("config.index"))
507 slug = _re.sub(r"[^a-z0-9]+", "-", raw).strip("-")[:64]
508 if not slug:
509 flash(_("Hangar ID must contain at least one letter or digit."), "danger")
510 return redirect(url_for("config.index"))
512 existing = Tenant.query.filter(Tenant.slug == slug, Tenant.id != tenant.id).first()
513 if existing:
514 flash(_("That Hangar ID is already in use. Please choose another."), "danger")
515 return redirect(url_for("config.index"))
517 old_slug = tenant.slug
518 tenant.slug = slug
520 if old_slug and old_slug != slug:
521 from documents.routes import _safe_join # pyright: ignore[reportMissingImports]
523 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
524 old_dir = _safe_join(folder, old_slug)
525 new_dir = _safe_join(folder, slug)
526 if os.path.isdir(old_dir):
527 if os.path.isdir(new_dir):
528 # Destination already exists — merge file-by-file
529 for dirpath, _dirs, filenames in os.walk(old_dir):
530 rel = os.path.relpath(dirpath, old_dir)
531 dest_dir = os.path.join(new_dir, rel)
532 os.makedirs(dest_dir, exist_ok=True)
533 for fname in filenames:
534 shutil.move(
535 os.path.join(dirpath, fname),
536 os.path.join(dest_dir, fname),
537 )
538 shutil.rmtree(old_dir, ignore_errors=True)
539 else:
540 os.rename(old_dir, new_dir)
542 # Rewrite stored paths in the database
543 prefix_old = old_slug + "/"
544 prefix_new = slug + "/"
545 for doc in Document.query.filter(Document.filename.like(old_slug + "/%")).all():
546 doc.filename = prefix_new + doc.filename[len(prefix_old) :]
547 for pr in PendingReconcile.query.filter(
548 PendingReconcile.filepath.like(old_slug + "/%")
549 ).all():
550 pr.filepath = prefix_new + pr.filepath[len(prefix_old) :]
551 for photo in AircraftPhoto.query.filter(
552 AircraftPhoto.filename.like(old_slug + "/%")
553 ).all():
554 photo.filename = prefix_new + photo.filename[len(prefix_old) :]
556 db.session.commit()
557 flash(_("Hangar ID saved."), "success")
558 return redirect(url_for("config.index"))
561@config_bp.route("/map-tiles", methods=["POST"])
562@login_required
563def update_map_tiles() -> ResponseReturnValue:
564 # ADMIN/OWNER enforcement is handled by config_bp.before_request.
565 key = request.form.get("openaip_api_key", "").strip()
566 setting = db.session.get(AppSetting, "openaip_api_key")
567 if key:
568 if setting:
569 setting.value = key
570 else:
571 db.session.add(AppSetting(key="openaip_api_key", value=key))
572 db.session.commit()
573 flash(_("OpenAIP API key saved."), "success")
574 else:
575 if setting:
576 db.session.delete(setting)
577 db.session.commit()
578 flash(_("OpenAIP API key removed."), "success")
579 return redirect(url_for("config.index"))
582@config_bp.route("/run", methods=["POST"])
583def run_backup_now() -> ResponseReturnValue:
584 if not session.get("user_id"):
585 abort(403)
586 try:
587 record = run_backup()
588 flash(_("Backup completed: %(filename)s", filename=record.filename), "success")
589 except RuntimeError as exc:
590 flash(_("Backup failed: %(error)s", error=exc), "danger")
591 return redirect(url_for("config.index"))
594@config_bp.route("/profile", methods=["POST"])
595def update_profile() -> ResponseReturnValue:
596 if not session.get("user_id"):
597 abort(403)
598 from models import ( # pyright: ignore[reportMissingImports]
599 OperatingModel,
600 Tenant,
601 TenantProfile,
602 TenantUser,
603 )
605 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
606 if not tu:
607 abort(403) # pragma: no cover
609 profile = TenantProfile.query.filter_by(tenant_id=tu.tenant_id).first()
610 if not profile:
611 profile = TenantProfile(tenant_id=tu.tenant_id, setup_complete=True)
612 db.session.add(profile)
614 model_str = request.form.get("operating_model", "sole_operator")
615 try:
616 profile.operating_model = OperatingModel(model_str)
617 except ValueError:
618 flash(_("Invalid operating model."), "danger")
619 return redirect(url_for("config.index"))
620 if model_str == "sole_pilot":
621 profile.planned_aircraft_count = 0
622 profile.allows_rental = False
623 else:
624 try:
625 count = max(1, int(request.form.get("planned_aircraft_count") or 1))
626 except (ValueError, TypeError):
627 count = 1
628 profile.planned_aircraft_count = count
629 profile.allows_rental = bool(request.form.get("allows_rental"))
631 policy = request.form.get("rental_authorization_policy", "").strip()
632 if policy in ("off", "warn", "block"):
633 profile.rental_authorization_policy = policy
635 if profile.operating_model == OperatingModel.SHARED_OWNERSHIP:
636 overdue_raw = request.form.get("co_owner_overdue_days", "").strip()
637 try:
638 overdue_days = int(overdue_raw)
639 if overdue_days >= 1:
640 profile.co_owner_overdue_days = overdue_days
641 except ValueError:
642 pass # non-numeric input — keep the existing setting unchanged
644 tenant = db.session.get(Tenant, tu.tenant_id)
645 if tenant:
646 tenant.require_totp = bool(request.form.get("require_totp"))
648 db.session.commit()
649 flash(_("Usage profile updated."), "success")
650 return redirect(url_for("config.index"))
653@config_bp.route("/email/test", methods=["POST"])
654def test_email() -> ResponseReturnValue:
655 if not session.get("user_id"):
656 abort(403)
657 from models import User # pyright: ignore[reportMissingImports]
658 from services.email_service import (
659 EmailNotConfiguredError,
660 EmailSendError,
661 send_email,
662 ) # pyright: ignore[reportMissingImports]
664 user = db.session.get(User, session["user_id"])
665 if not user:
666 abort(403) # pragma: no cover
667 try:
668 send_email(
669 to=user.email,
670 subject="OpenHangar — test email",
671 text_body=(
672 "This is a test email from your OpenHangar instance.\n\n"
673 "If you received this, your SMTP configuration is working correctly."
674 ),
675 )
676 flash(_("Test email sent to %(email)s.", email=user.email), "success")
677 except EmailNotConfiguredError as exc:
678 flash(_("Email not configured: %(error)s", error=exc), "warning")
679 except EmailSendError as exc:
680 flash(_("Email send failed: %(error)s", error=exc), "danger")
681 return redirect(url_for("config.index"))
684@config_bp.route("/check-version", methods=["POST"])
685def check_version() -> ResponseReturnValue:
686 if not session.get("user_id"):
687 abort(403)
688 import json as _json
689 from datetime import datetime
691 from services.version_service import ( # pyright: ignore[reportMissingImports]
692 fetch_versions,
693 upsert_app_setting,
694 )
696 versions = fetch_versions()
697 upsert_app_setting(
698 db.session,
699 "version_last_checked_at",
700 datetime.now(UTC).isoformat(),
701 )
702 if versions:
703 upsert_app_setting(db.session, "latest_version", versions[0])
704 upsert_app_setting(db.session, "all_versions", _json.dumps(versions))
705 from services.version_service import (
706 _persist_update_flag, # pyright: ignore[reportMissingImports]
707 )
709 _persist_update_flag(
710 db.session, os.environ.get("OPENHANGAR_VERSION", "development"), versions[0]
711 )
712 db.session.commit()
713 flash(_("Version check refreshed."), "success")
714 return redirect(url_for("config.index"))
717# ── One-click upgrade ─────────────────────────────────────────────────────────
720def _upgrade_trigger_state(upgrade_dir: str) -> str:
721 """Returns "running", "triggered" or "idle" — shared by the manual button
722 and the scheduled auto-upgrade check (services/backup_scheduler.py) so
723 both agree on whether it's safe to write a new trigger file."""
724 os.makedirs(upgrade_dir, exist_ok=True)
725 if os.path.exists(os.path.join(upgrade_dir, "trigger.running")):
726 return "running"
727 if os.path.exists(os.path.join(upgrade_dir, "trigger")):
728 return "triggered"
729 return "idle"
732def _write_upgrade_trigger(upgrade_dir: str, triggered_by: str) -> None:
733 trigger_data = {
734 "triggered_by": triggered_by,
735 "triggered_at": datetime.now(UTC).isoformat(),
736 }
737 with open(os.path.join(upgrade_dir, "trigger"), "w") as fh:
738 json.dump(trigger_data, fh)
741def _record_last_upgrade(upgrade_dir: str) -> bool:
742 """One-shot: if docker/upgrade.sh left a trigger.done marker (written
743 right after a successful pull + container recreate — see its own
744 ``"ok $(date -u +%FT%TZ)" > "${DONE}"`` line), persist that timestamp as
745 "last upgrade" and remove the marker. Returns True if a marker was found
746 (and thus consumed) so callers building the live-poll JSON response
747 don't have to re-check file existence themselves.
749 Called from both /config's own page load (catches an unattended
750 auto-upgrade completing overnight, where no browser was ever polling)
751 and /upgrade-status's live poll (catches the normal attended case, which
752 usually wins the race since it polls every couple of seconds right as
753 the upgrade finishes) — whichever sees the marker first "wins" and the
754 other finds it already gone, the same one-shot idiom as EE-10's per-day
755 session flag in init.py.
756 """
757 done_path = os.path.join(upgrade_dir, "trigger.done")
758 if not os.path.exists(done_path):
759 return False
760 content = ""
761 with contextlib.suppress(OSError):
762 with open(done_path) as fh:
763 content = fh.read().strip()
764 os.remove(done_path)
765 # Content is "ok <ISO8601Z>" — fall back to "now" if that ever doesn't
766 # parse (e.g. a stale marker from an older upgrade.sh revision).
767 parts = content.split()
768 ts = parts[1] if len(parts) > 1 else datetime.now(UTC).isoformat()
769 from services.version_service import ( # pyright: ignore[reportMissingImports]
770 upsert_app_setting,
771 )
773 upsert_app_setting(db.session, "last_upgrade_at", ts)
774 db.session.commit()
775 return True
778@config_bp.route("/trigger-upgrade", methods=["POST"])
779@require_instance_admin
780def trigger_upgrade() -> ResponseReturnValue:
781 upgrade_dir = os.environ.get("OPENHANGAR_UPGRADE_DIR", "")
782 if not upgrade_dir:
783 abort(404)
784 state = _upgrade_trigger_state(upgrade_dir)
785 if state == "running":
786 flash(_("An upgrade is already in progress."), "warning")
787 return redirect(url_for("config.index"))
788 if state == "triggered":
789 flash(_("Upgrade already triggered."), "info")
790 return redirect(url_for("config.index"))
792 # Guarantee a fallback exists before handing off to the external upgrade
793 # process: once the trigger file is written, it may be picked up at any
794 # time, so the backup must complete first. Block the upgrade if it fails
795 # rather than proceed without a fresh safety net.
796 try:
797 record = run_backup()
798 except RuntimeError as exc:
799 flash(
800 _(
801 "Pre-upgrade backup failed, upgrade not triggered: %(error)s",
802 error=exc,
803 ),
804 "danger",
805 )
806 return redirect(url_for("config.index"))
808 from models import User # pyright: ignore[reportMissingImports]
810 user = db.session.get(User, session["user_id"])
811 _write_upgrade_trigger(upgrade_dir, user.email if user else "unknown")
812 flash(
813 _(
814 "Backup created (%(filename)s). Upgrade triggered — the service "
815 "will restart shortly.",
816 filename=record.filename,
817 ),
818 "info",
819 )
820 return redirect(url_for("config.index"))
823@config_bp.route("/upgrade-status")
824@require_instance_admin
825def upgrade_status() -> ResponseReturnValue:
826 upgrade_dir = os.environ.get("OPENHANGAR_UPGRADE_DIR", "")
827 if not upgrade_dir:
828 return abort(404)
829 # Included in every response below so the client can detect completion by
830 # comparing against the version it saw at page load, independent of the
831 # trigger.* file flags — those live on a bind-mount shared by the outgoing
832 # and incoming container across the handover, so a poll can land on either
833 # one and the one-shot "done" file can be consumed by a request other than
834 # the one deciding whether to reload.
835 version = os.environ.get("OPENHANGAR_VERSION", "development")
836 failed_path = os.path.join(upgrade_dir, "trigger.failed")
837 running_path = os.path.join(upgrade_dir, "trigger.running")
838 trigger_path = os.path.join(upgrade_dir, "trigger")
839 if _record_last_upgrade(upgrade_dir):
840 return jsonify({"status": "done", "version": version})
841 if os.path.exists(failed_path):
842 msg = ""
843 with contextlib.suppress(OSError):
844 with open(failed_path) as fh:
845 msg = fh.read().strip()
846 os.remove(failed_path)
847 return jsonify({"status": "failed", "message": msg, "version": version})
848 if os.path.exists(running_path):
849 return jsonify({"status": "in-progress", "version": version})
850 if os.path.exists(trigger_path):
851 return jsonify({"status": "triggered", "version": version})
852 return jsonify({"status": "idle", "version": version})
855# ── Phase 29: Tenant management (instance admin only) ─────────────────────────
858@config_bp.route("/tenants")
859@require_instance_admin
860def tenant_list() -> ResponseReturnValue:
861 from models import ( # pyright: ignore[reportMissingImports]
862 Aircraft,
863 Role,
864 Tenant,
865 TenantUser,
866 )
868 tenants = Tenant.query.order_by(Tenant.created_at).all()
869 stats = []
870 for t in tenants:
871 user_count = TenantUser.query.filter_by(tenant_id=t.id).count()
872 aircraft_count = Aircraft.query.filter_by(tenant_id=t.id).count()
873 owners = (
874 TenantUser.query.filter_by(tenant_id=t.id)
875 .filter(TenantUser.role.in_([Role.OWNER, Role.ADMIN]))
876 .all()
877 )
878 stats.append(
879 {
880 "tenant": t,
881 "user_count": user_count,
882 "aircraft_count": aircraft_count,
883 "owners": owners,
884 }
885 )
887 return render_template("config/tenant_list.html", stats=stats)
890@config_bp.route("/tenants/create", methods=["GET", "POST"])
891@require_instance_admin
892def tenant_create() -> ResponseReturnValue:
893 from datetime import timedelta
895 from models import ( # pyright: ignore[reportMissingImports]
896 OperatingModel,
897 Role,
898 Tenant,
899 TenantProfile,
900 User,
901 UserInvitation,
902 )
904 user = db.session.get(User, session["user_id"])
905 assert user is not None # guaranteed by @require_instance_admin
907 if request.method == "POST":
908 name = request.form.get("name", "").strip()
909 admin_email = request.form.get("admin_email", "").strip().lower()
910 model_str = request.form.get("operating_model", "sole_operator")
912 if not name:
913 flash(_("Tenant name is required."), "danger")
914 return render_template("config/tenant_create.html")
915 if not admin_email:
916 flash(_("Admin email is required."), "danger")
917 return render_template("config/tenant_create.html")
919 tenant = Tenant(name=name, is_active=True)
920 db.session.add(tenant)
921 db.session.flush()
923 try:
924 op_model: OperatingModel | None = OperatingModel(model_str)
925 except ValueError:
926 op_model = None
928 profile = TenantProfile(
929 tenant_id=tenant.id,
930 operating_model=op_model,
931 setup_complete=False,
932 )
933 db.session.add(profile)
935 invitation = UserInvitation(
936 tenant_id=tenant.id,
937 invited_by_user_id=user.id,
938 email=admin_email,
939 role=Role.OWNER,
940 expires_at=datetime.now(UTC) + timedelta(days=7),
941 )
942 db.session.add(invitation)
943 db.session.commit()
945 flash(
946 _(
947 "Tenant '%(name)s' created. Share this invite link with the owner: %(url)s",
948 name=name,
949 url=url_for(
950 "users.accept_invite", token=invitation.token, _external=True
951 ),
952 ),
953 "success",
954 )
955 return redirect(url_for("config.tenant_list"))
957 return render_template("config/tenant_create.html")
960@config_bp.route("/tenants/<int:tenant_id>/toggle", methods=["POST"])
961@require_instance_admin
962def tenant_toggle_active(tenant_id: int) -> ResponseReturnValue:
963 from models import Tenant # pyright: ignore[reportMissingImports]
965 tenant = db.session.get(Tenant, tenant_id)
966 if not tenant:
967 abort(404)
969 tenant.is_active = not tenant.is_active
970 db.session.commit()
972 if tenant.is_active:
973 flash(_("Tenant '%(name)s' reactivated.", name=tenant.name), "success")
974 else:
975 flash(_("Tenant '%(name)s' deactivated.", name=tenant.name), "warning")
977 return redirect(url_for("config.tenant_list"))
980@config_bp.route("/tenants/<int:tenant_id>/reset-password", methods=["POST"])
981@require_instance_admin
982def tenant_reset_owner_password(tenant_id: int) -> ResponseReturnValue:
983 from datetime import timedelta
985 from models import ( # pyright: ignore[reportMissingImports]
986 PasswordResetToken,
987 Role,
988 Tenant,
989 TenantUser,
990 User,
991 )
993 admin = db.session.get(User, session["user_id"])
994 assert admin is not None # guaranteed by @require_instance_admin
995 tenant = db.session.get(Tenant, tenant_id)
996 if not tenant:
997 abort(404)
999 owner_user_id = request.form.get("owner_user_id", type=int)
1000 if not owner_user_id:
1001 flash(_("No user selected."), "danger")
1002 return redirect(url_for("config.tenant_list"))
1004 tu = TenantUser.query.filter_by(tenant_id=tenant_id, user_id=owner_user_id).first()
1005 if not tu or tu.role not in (Role.OWNER, Role.ADMIN):
1006 abort(403)
1008 token = PasswordResetToken(
1009 user_id=owner_user_id,
1010 generated_by_user_id=admin.id,
1011 expires_at=datetime.now(UTC) + timedelta(hours=24),
1012 )
1013 db.session.add(token)
1014 db.session.commit()
1016 reset_url = url_for("auth.reset_password", token=token.token, _external=True)
1017 return render_template(
1018 "config/tenant_reset_token.html",
1019 tenant=tenant,
1020 reset_url=reset_url,
1021 expires_at=token.expires_at,
1022 )
1025@config_bp.route("/notifications/", methods=["GET", "POST"])
1026@login_required
1027def notification_preferences() -> ResponseReturnValue:
1028 """Manage per-user notification preferences. Accessible to all logged-in users."""
1029 from models import ( # pyright: ignore[reportMissingImports]
1030 NotificationPreference,
1031 NotificationType,
1032 Role,
1033 TenantNotificationDefault,
1034 TenantProfile,
1035 TenantUser,
1036 User,
1037 )
1038 from utils import current_user_role # pyright: ignore[reportMissingImports]
1040 user = db.session.get(User, session["user_id"])
1041 if not user:
1042 abort(403)
1044 tu = TenantUser.query.filter_by(user_id=user.id).first()
1045 tenant_id = tu.tenant_id if tu else None
1046 role = current_user_role()
1048 is_owner = role in (Role.ADMIN, Role.OWNER)
1049 is_pilot = (
1050 role in (Role.ADMIN, Role.OWNER, Role.PILOT, Role.INSTRUCTOR) or user.is_pilot
1051 )
1052 is_maint = (
1053 role in (Role.ADMIN, Role.OWNER, Role.MAINTENANCE, Role.INSTRUCTOR)
1054 or user.is_maintenance
1055 )
1057 def _user_has_cap(caps: list[str]) -> bool:
1058 return (
1059 ("is_owner" in caps and is_owner)
1060 or ("is_pilot" in caps and is_pilot)
1061 or ("is_maint" in caps and is_maint)
1062 )
1064 visible_types = [
1065 t
1066 for t in NotificationType.ALL
1067 if _user_has_cap(NotificationType.REQUIRED_CAPS.get(t, []))
1068 ]
1070 if request.method == "POST":
1071 if tenant_id is None:
1072 flash(_("Cannot save: no tenant associated."), "danger")
1073 return redirect(url_for("config.notification_preferences"))
1075 for notif_type in visible_types:
1076 enabled = bool(request.form.get(f"enabled_{notif_type}"))
1077 threshold_raw = request.form.get(f"threshold_{notif_type}", "").strip()
1078 threshold_days: int | None = None
1079 if notif_type in NotificationType.HAS_THRESHOLD and threshold_raw:
1080 try:
1081 threshold_days = max(1, int(threshold_raw))
1082 except ValueError:
1083 threshold_days = None
1085 existing = NotificationPreference.query.filter_by(
1086 user_id=user.id, tenant_id=tenant_id, notification_type=notif_type
1087 ).first()
1088 # Only save if the user's preference differs from the effective default
1089 system_default = NotificationType.SYSTEM_DEFAULTS.get(notif_type, {})
1090 same_as_default = enabled == system_default.get(
1091 "enabled", False
1092 ) and threshold_days == system_default.get("threshold_days")
1093 if same_as_default and existing:
1094 db.session.delete(existing)
1095 elif not same_as_default:
1096 if existing:
1097 existing.enabled = enabled
1098 existing.threshold_days = threshold_days
1099 else:
1100 db.session.add(
1101 NotificationPreference(
1102 user_id=user.id,
1103 tenant_id=tenant_id,
1104 notification_type=notif_type,
1105 enabled=enabled,
1106 threshold_days=threshold_days,
1107 )
1108 )
1109 db.session.commit()
1110 flash(_("Notification preferences saved."), "success")
1111 return redirect(url_for("config.notification_preferences"))
1113 # Build current effective preferences for display
1114 prefs: dict[str, dict[str, object]] = {}
1115 for notif_type in visible_types:
1116 if tenant_id:
1117 from services.notification_service import (
1118 get_effective_preference, # pyright: ignore[reportMissingImports]
1119 )
1121 prefs[notif_type] = get_effective_preference(user.id, tenant_id, notif_type)
1122 else:
1123 prefs[notif_type] = dict(
1124 NotificationType.SYSTEM_DEFAULTS.get(
1125 notif_type, {"enabled": False, "threshold_days": None}
1126 )
1127 )
1129 # Tenant defaults visible only to admins/owners
1130 tenant_defaults: dict[str, dict[str, object]] | None = None
1131 if is_owner and tenant_id:
1132 tenant_defaults = {}
1133 for notif_type in NotificationType.ALL:
1134 td = TenantNotificationDefault.query.filter_by(
1135 tenant_id=tenant_id, notification_type=notif_type
1136 ).first()
1137 if td:
1138 tenant_defaults[notif_type] = {
1139 "enabled": td.enabled,
1140 "threshold_days": td.threshold_days,
1141 }
1142 else:
1143 tenant_defaults[notif_type] = dict(
1144 NotificationType.SYSTEM_DEFAULTS.get(
1145 notif_type, {"enabled": False, "threshold_days": None}
1146 )
1147 )
1149 profile = (
1150 TenantProfile.query.filter_by(tenant_id=tenant_id).first()
1151 if tenant_id
1152 else None
1153 )
1155 return render_template(
1156 "config/notifications.html",
1157 visible_types=visible_types,
1158 prefs=prefs,
1159 has_threshold=NotificationType.HAS_THRESHOLD,
1160 system_defaults=NotificationType.SYSTEM_DEFAULTS,
1161 tenant_defaults=tenant_defaults,
1162 is_owner=is_owner,
1163 profile=profile,
1164 )
1167@config_bp.route("/backfill/aircraft-type-icao", methods=["POST"])
1168@require_instance_admin
1169def backfill_aircraft_type_icao() -> ResponseReturnValue:
1170 """Resolve other_aircraft_type_icao for all standalone Flight rows that
1171 have an other_aircraft_type but no icao designator."""
1172 from models import Flight # pyright: ignore[reportMissingImports]
1173 from utils import (
1174 resolve_aircraft_type_icao, # pyright: ignore[reportMissingImports]
1175 )
1177 rows = Flight.query.filter(
1178 Flight.aircraft_id.is_(None),
1179 Flight.other_aircraft_type.isnot(None),
1180 Flight.other_aircraft_type_icao.is_(None),
1181 ).all()
1183 updated = 0
1184 for entry in rows:
1185 resolved = resolve_aircraft_type_icao(entry.other_aircraft_type)
1186 if resolved:
1187 entry.other_aircraft_type_icao = resolved
1188 updated += 1
1190 db.session.commit()
1191 flash(
1192 ngettext(
1193 "Back-fill complete: one of %(total)d entry resolved.",
1194 "Back-fill complete: %(updated)d of %(total)d entries resolved.",
1195 updated,
1196 updated=updated,
1197 total=len(rows),
1198 ),
1199 "success",
1200 )
1201 return redirect(url_for("config.index"))
1204@config_bp.route("/backfill/pilot-log-to-flight-entries", methods=["POST"])
1205@require_instance_admin
1206def backfill_pilot_log_to_flight_entries() -> ResponseReturnValue:
1207 """Promote standalone Flight rows (aircraft_id NULL, other_aircraft_*
1208 registration set) to a managed aircraft, for any whose registration
1209 now matches one."""
1210 from models import Flight # pyright: ignore[reportMissingImports]
1211 from pilots.logbook_import import (
1212 link_entries_to_aircraft, # pyright: ignore[reportMissingImports]
1213 )
1215 entries = (
1216 Flight.query.filter(
1217 Flight.aircraft_id.is_(None),
1218 Flight.other_aircraft_registration.isnot(None),
1219 )
1220 .order_by(Flight.date.asc(), Flight.id.asc())
1221 .all()
1222 )
1224 created = link_entries_to_aircraft(entries)
1225 db.session.commit()
1227 flash(
1228 ngettext(
1229 "Back-fill complete: %(count)d entry linked to a managed aircraft.",
1230 "Back-fill complete: %(count)d entries linked to a managed aircraft.",
1231 created,
1232 count=created,
1233 ),
1234 "success",
1235 )
1236 return redirect(url_for("config.index"))
1239_ALLOWED_BADGE_PATHS: dict[str, str] = {
1240 "uptimes/1h/badge.svg": "uptimes/1h/badge.svg",
1241 "uptimes/24h/badge.svg": "uptimes/24h/badge.svg",
1242 "uptimes/7d/badge.svg": "uptimes/7d/badge.svg",
1243 "uptimes/30d/badge.svg": "uptimes/30d/badge.svg",
1244 "response-times/1h/badge.svg": "response-times/1h/badge.svg",
1245 "response-times/24h/badge.svg": "response-times/24h/badge.svg",
1246 "response-times/7d/badge.svg": "response-times/7d/badge.svg",
1247 "response-times/30d/badge.svg": "response-times/30d/badge.svg",
1248}
1251# ── Renter authorizations (Phase 37c) ───────────────────────────────────────────
1254def _renter_auth_status(auth: Any) -> str:
1255 """'revoked' | 'expired' | 'expiring' | 'valid' — for the list badge."""
1256 from datetime import date as _date
1257 from datetime import timedelta as _timedelta
1259 if auth.revoked_at is not None:
1260 return "revoked"
1261 if not auth.is_valid:
1262 return "expired"
1263 today = _date.today()
1264 soon = today + _timedelta(days=30)
1265 dates = [d for d in (auth.expires_on, auth.medical_valid_until) if d is not None]
1266 if dates and min(dates) <= soon:
1267 return "expiring"
1268 return "valid"
1271def _get_renter_auth_or_404(tenant_id: int, auth_id: int) -> Any:
1272 from models import RenterAuthorization # pyright: ignore[reportMissingImports]
1274 auth = db.session.get(RenterAuthorization, auth_id)
1275 if not auth or auth.tenant_id != tenant_id:
1276 abort(404)
1277 return auth
1280@config_bp.route("/renters/")
1281@login_required
1282def renters_list() -> ResponseReturnValue:
1283 from models import ( # pyright: ignore[reportMissingImports]
1284 RenterAuthorization,
1285 TenantUser,
1286 )
1288 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1289 if not tu:
1290 abort(403) # pragma: no cover
1292 authorizations = (
1293 RenterAuthorization.query.filter_by(tenant_id=tu.tenant_id)
1294 .order_by(RenterAuthorization.granted_on.desc())
1295 .all()
1296 )
1297 rows = [(auth, _renter_auth_status(auth)) for auth in authorizations]
1298 return render_template("config/renters_list.html", rows=rows)
1301def _renter_auth_form_context(tenant_id: int) -> dict[str, Any]:
1302 from models import ( # pyright: ignore[reportMissingImports]
1303 Aircraft,
1304 TenantUser,
1305 User,
1306 )
1308 tenant_users = (
1309 TenantUser.query.filter_by(tenant_id=tenant_id)
1310 .join(User)
1311 .order_by(User.email)
1312 .all()
1313 )
1314 aircraft_list = (
1315 Aircraft.query.filter_by(tenant_id=tenant_id, archived_at=None)
1316 .order_by(Aircraft.registration)
1317 .all()
1318 )
1319 return {"tenant_users": tenant_users, "aircraft_list": aircraft_list}
1322def _save_renter_authorization(tenant_id: int, auth: Any | None) -> ResponseReturnValue:
1323 from datetime import date as _date
1325 from models import ( # pyright: ignore[reportMissingImports]
1326 Aircraft,
1327 RenterAuthorization,
1328 TenantUser,
1329 )
1331 def _date_or_none(key: str) -> _date | None:
1332 raw = request.form.get(key, "").strip()
1333 if not raw:
1334 return None
1335 try:
1336 return _date.fromisoformat(raw)
1337 except ValueError:
1338 return None
1340 renter_user_id_raw = request.form.get("renter_user_id", "").strip()
1341 aircraft_id_raw = request.form.get("aircraft_id", "").strip()
1342 granted_on = _date_or_none("granted_on")
1343 expires_on = _date_or_none("expires_on")
1344 checkout_flight_on = _date_or_none("checkout_flight_on")
1345 licence_seen_on = _date_or_none("licence_seen_on")
1346 medical_valid_until = _date_or_none("medical_valid_until")
1347 notes = request.form.get("notes", "").strip() or None
1349 errors = []
1350 renter_user_id: int | None = None
1351 if renter_user_id_raw:
1352 try:
1353 renter_user_id = int(renter_user_id_raw)
1354 except ValueError:
1355 renter_user_id = None
1356 if (
1357 renter_user_id is None
1358 or not TenantUser.query.filter_by(
1359 tenant_id=tenant_id, user_id=renter_user_id
1360 ).first()
1361 ):
1362 errors.append(_("Select a valid renter."))
1364 aircraft_id: int | None = None
1365 if aircraft_id_raw:
1366 try:
1367 aircraft_id = int(aircraft_id_raw)
1368 except ValueError:
1369 errors.append(_("Invalid aircraft selection."))
1370 else:
1371 if not Aircraft.query.filter_by(
1372 id=aircraft_id, tenant_id=tenant_id
1373 ).first():
1374 errors.append(_("Invalid aircraft selection."))
1376 if granted_on is None:
1377 errors.append(_("Granted date is required and must be a valid date."))
1378 if expires_on is not None and granted_on is not None and expires_on < granted_on:
1379 errors.append(_("Expiry date cannot be before the granted date."))
1381 agreement_file = request.files.get("agreement")
1382 if agreement_file is not None and not agreement_file.filename:
1383 agreement_file = None
1384 if agreement_file is not None:
1385 import os as _os
1387 from documents.routes import (
1388 _ALLOWED_EXTS, # pyright: ignore[reportMissingImports]
1389 )
1391 ext = _os.path.splitext(agreement_file.filename or "")[1].lower()
1392 if ext not in _ALLOWED_EXTS:
1393 errors.append(_("This file type is not allowed for the rental agreement."))
1395 if errors:
1396 for msg in errors:
1397 flash(msg, "danger")
1398 ctx = _renter_auth_form_context(tenant_id)
1399 return render_template("config/renter_form.html", auth=auth, **ctx)
1401 if auth is None:
1402 auth = RenterAuthorization(
1403 tenant_id=tenant_id,
1404 authorized_by_id=session["user_id"],
1405 )
1406 db.session.add(auth)
1408 auth.renter_user_id = renter_user_id
1409 auth.aircraft_id = aircraft_id
1410 auth.granted_on = granted_on
1411 auth.expires_on = expires_on
1412 auth.checkout_flight_on = checkout_flight_on
1413 auth.licence_seen_on = licence_seen_on
1414 auth.medical_valid_until = medical_valid_until
1415 auth.notes = notes
1417 if agreement_file is not None:
1418 from documents.routes import (
1419 _save_upload, # pyright: ignore[reportMissingImports]
1420 )
1421 from models import Document # pyright: ignore[reportMissingImports]
1422 from werkzeug.utils import (
1423 secure_filename, # pyright: ignore[reportMissingImports]
1424 )
1426 db.session.flush()
1427 stored, mime, size = _save_upload(agreement_file, f"renter-agreement-{auth.id}")
1428 db.session.add(
1429 Document(
1430 renter_authorization_id=auth.id,
1431 filename=stored,
1432 original_filename=secure_filename(
1433 agreement_file.filename or "agreement"
1434 ),
1435 mime_type=mime,
1436 size_bytes=size,
1437 title=_("Rental agreement"),
1438 is_sensitive=True,
1439 )
1440 )
1442 db.session.commit()
1443 flash(_("Renter authorization saved."), "success")
1444 return redirect(url_for("config.renters_list"))
1447@config_bp.route("/renters/add", methods=["GET", "POST"])
1448@login_required
1449def renter_add() -> ResponseReturnValue:
1450 from models import TenantUser # pyright: ignore[reportMissingImports]
1452 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1453 if not tu:
1454 abort(403) # pragma: no cover
1456 if request.method == "POST":
1457 return _save_renter_authorization(tu.tenant_id, None)
1459 ctx = _renter_auth_form_context(tu.tenant_id)
1460 return render_template("config/renter_form.html", auth=None, **ctx)
1463@config_bp.route("/renters/<int:auth_id>/edit", methods=["GET", "POST"])
1464@login_required
1465def renter_edit(auth_id: int) -> ResponseReturnValue:
1466 from models import TenantUser # pyright: ignore[reportMissingImports]
1468 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1469 if not tu:
1470 abort(403) # pragma: no cover
1471 auth = _get_renter_auth_or_404(tu.tenant_id, auth_id)
1473 if request.method == "POST":
1474 return _save_renter_authorization(tu.tenant_id, auth)
1476 ctx = _renter_auth_form_context(tu.tenant_id)
1477 return render_template("config/renter_form.html", auth=auth, **ctx)
1480@config_bp.route("/renters/<int:auth_id>/revoke", methods=["POST"])
1481@login_required
1482def renter_revoke(auth_id: int) -> ResponseReturnValue:
1483 from models import TenantUser # pyright: ignore[reportMissingImports]
1485 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1486 if not tu:
1487 abort(403) # pragma: no cover
1488 auth = _get_renter_auth_or_404(tu.tenant_id, auth_id)
1490 auth.revoked_at = datetime.now(UTC)
1491 db.session.commit()
1492 flash(_("Renter authorization revoked."), "success")
1493 return redirect(url_for("config.renters_list"))
1496# ── Renter billing account (Phase 37e) ──────────────────────────────────────────
1499def _renter_account_period(period_months_raw: str | None) -> tuple[Any, Any]:
1500 from datetime import date, timedelta
1502 try:
1503 period_months = int(period_months_raw) if period_months_raw else 12
1504 except ValueError:
1505 period_months = 12
1506 if period_months <= 0:
1507 period_months = 12
1508 end = date.today()
1509 start = end - timedelta(days=period_months * 30)
1510 return start, end
1513@config_bp.route("/renters/<int:user_id>/account")
1514@login_required
1515def renter_account(user_id: int) -> ResponseReturnValue:
1516 from models import ( # pyright: ignore[reportMissingImports]
1517 BillingAccountKind,
1518 TenantUser,
1519 User,
1520 )
1521 from services.billing import BillingService # pyright: ignore[reportMissingImports]
1523 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1524 if not tu:
1525 abort(403) # pragma: no cover
1526 renter_tu = TenantUser.query.filter_by(
1527 user_id=user_id, tenant_id=tu.tenant_id
1528 ).first()
1529 if not renter_tu:
1530 abort(404)
1531 renter = db.session.get(User, user_id)
1532 if renter is None: # pragma: no cover — FK guarantees it exists
1533 abort(404)
1535 account = BillingService.get_or_create_account(
1536 tu.tenant_id, user_id, BillingAccountKind.RENTER
1537 )
1538 db.session.commit() # persist a newly-created account
1539 start, end = _renter_account_period(request.args.get("period"))
1540 statement = BillingService.statement(account, start, end)
1542 return render_template(
1543 "config/renter_account.html",
1544 renter=renter,
1545 account=account,
1546 statement=statement,
1547 balance=BillingService.balance(account),
1548 is_owner_view=True,
1549 csv_url=url_for("config.renter_statement_csv", user_id=user_id),
1550 )
1553@config_bp.route("/renters/<int:user_id>/account/payment", methods=["POST"])
1554@login_required
1555def renter_record_payment(user_id: int) -> ResponseReturnValue:
1556 from models import ( # pyright: ignore[reportMissingImports]
1557 BillingAccountKind,
1558 TenantUser,
1559 )
1560 from services.billing import BillingService # pyright: ignore[reportMissingImports]
1562 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1563 if not tu:
1564 abort(403) # pragma: no cover
1565 renter_tu = TenantUser.query.filter_by(
1566 user_id=user_id, tenant_id=tu.tenant_id
1567 ).first()
1568 if not renter_tu:
1569 abort(404)
1571 from datetime import date as _date_cls
1573 amount_raw = request.form.get("amount", "").strip()
1574 date_raw = request.form.get("date", "").strip()
1575 note = request.form.get("note", "").strip() or None
1577 errors = []
1578 try:
1579 amount = float(amount_raw)
1580 if amount <= 0:
1581 errors.append(_("Payment amount must be positive."))
1582 except ValueError:
1583 amount = 0.0
1584 errors.append(_("Invalid payment amount."))
1585 try:
1586 payment_date = (
1587 _date_cls.fromisoformat(date_raw) if date_raw else _date_cls.today()
1588 )
1589 except ValueError:
1590 payment_date = _date_cls.today()
1591 errors.append(_("Invalid payment date."))
1593 if errors:
1594 for msg in errors:
1595 flash(msg, "danger")
1596 return redirect(url_for("config.renter_account", user_id=user_id))
1598 from models import LedgerEntryType, User # pyright: ignore[reportMissingImports]
1600 account = BillingService.get_or_create_account(
1601 tu.tenant_id, user_id, BillingAccountKind.RENTER
1602 )
1603 recorder = db.session.get(User, session["user_id"])
1604 BillingService.post(
1605 account,
1606 LedgerEntryType.PAYMENT,
1607 -amount,
1608 note or str(_("Payment received")),
1609 payment_date,
1610 source_type="payment",
1611 created_by=recorder,
1612 )
1613 db.session.commit()
1614 flash(_("Payment recorded."), "success")
1615 return redirect(url_for("config.renter_account", user_id=user_id))
1618@config_bp.route("/renters/<int:user_id>/account/statement.csv")
1619@login_required
1620def renter_statement_csv(user_id: int) -> ResponseReturnValue:
1621 from flask import Response # pyright: ignore[reportMissingImports]
1622 from models import ( # pyright: ignore[reportMissingImports]
1623 BillingAccountKind,
1624 TenantUser,
1625 User,
1626 )
1627 from services.billing import BillingService # pyright: ignore[reportMissingImports]
1629 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1630 if not tu:
1631 abort(403) # pragma: no cover
1632 renter_tu = TenantUser.query.filter_by(
1633 user_id=user_id, tenant_id=tu.tenant_id
1634 ).first()
1635 if not renter_tu:
1636 abort(404)
1638 account = BillingService.get_or_create_account(
1639 tu.tenant_id, user_id, BillingAccountKind.RENTER
1640 )
1641 db.session.commit()
1642 start, end = _renter_account_period(request.args.get("period"))
1643 statement = BillingService.statement(account, start, end)
1644 exporter = db.session.get(User, session["user_id"])
1645 csv_text = BillingService.statement_csv(statement, exported_by=exporter)
1646 return Response(
1647 csv_text,
1648 mimetype="text/csv",
1649 headers={
1650 "Content-Disposition": f"attachment; filename=statement_{user_id}_{start.isoformat()}_{end.isoformat()}.csv"
1651 },
1652 )
1655@config_bp.route("/gatus-badge/<path:badge_path>")
1656@login_required
1657def gatus_badge(badge_path: str) -> ResponseReturnValue:
1658 safe_path = _ALLOWED_BADGE_PATHS.get(badge_path)
1659 if safe_path is None:
1660 return abort(404)
1661 gatus = _parse_gatus_env()
1662 if gatus is None:
1663 return abort(404)
1664 base_url, endpoint_key, auth_header = gatus
1665 badge_url = f"{base_url}/api/v1/endpoints/{endpoint_key}/{safe_path}"
1666 req = urllib.request.Request(badge_url)
1667 if auth_header:
1668 req.add_header("Authorization", f"Basic {auth_header}")
1669 try:
1670 with urllib.request.urlopen(req, timeout=5) as resp: # nosec B310
1671 content = resp.read()
1672 content_type = resp.headers.get("Content-Type", "image/svg+xml")
1673 return current_app.response_class(content, mimetype=content_type)
1674 except urllib.error.URLError as exc:
1675 log.warning("Gatus badge fetch failed (%s): %s", badge_url, repr(exc))
1676 return abort(503)