Coverage for app/config/routes.py: 100%
837 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +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_backup_keep,
273 parse_backup_keep_days,
274 parse_backup_keep_months,
275 parse_backup_keep_weeks,
276 parse_backup_retention,
277 parse_backup_time,
278 )
280 try:
281 _schedule = parse_backup_time()
282 except ValueError:
283 _schedule = None
284 backup_schedule_str = (
285 f"{_schedule[0]:02d}:{_schedule[1]:02d}" if _schedule else None
286 )
287 try:
288 backup_keep = parse_backup_keep()
289 except ValueError:
290 backup_keep = None
291 backup_gfs = None
292 try:
293 if parse_backup_retention() == RETENTION_GFS:
294 backup_gfs = {
295 "days": parse_backup_keep_days(),
296 "weeks": parse_backup_keep_weeks(),
297 "months": parse_backup_keep_months(),
298 }
299 except ValueError:
300 backup_gfs = None
301 _last_ok = (
302 BackupRecord.query.filter_by(status="ok")
303 .order_by(BackupRecord.created_at.desc())
304 .first()
305 )
306 last_backup_at = _last_ok.created_at if _last_ok else None
307 if last_backup_at is not None and last_backup_at.tzinfo is None:
308 # SQLite returns naive datetimes; values are stored in UTC.
309 last_backup_at = last_backup_at.replace(tzinfo=UTC)
310 backup_stale = backup_schedule_str is not None and (
311 last_backup_at is None
312 or datetime.now(UTC) - last_backup_at > _timedelta(days=2)
313 )
315 from models import ( # pyright: ignore[reportMissingImports]
316 Role,
317 TenantUser,
318 User,
319 UserInvitation,
320 )
321 from sqlalchemy import func # pyright: ignore[reportMissingImports]
323 _role_labels = {
324 Role.ADMIN: "Admin",
325 Role.OWNER: "Owner",
326 Role.PILOT: "Pilot / Renter",
327 Role.MAINTENANCE: "Maintenance",
328 Role.VIEWER: "Viewer",
329 }
330 tu_self = TenantUser.query.filter_by(user_id=session["user_id"]).first()
331 tid = tu_self.tenant_id if tu_self else None
332 user_counts = []
333 open_invitations = 0
334 if tid:
335 results = (
336 db.session.query(TenantUser.role, func.count(TenantUser.user_id))
337 .join(User, TenantUser.user_id == User.id)
338 .filter(TenantUser.tenant_id == tid, User.is_active.is_(True))
339 .group_by(TenantUser.role)
340 .all()
341 )
342 counts_by_role = dict(results)
343 user_counts = [
344 (_role_labels[r], counts_by_role[r])
345 for r in Role
346 if counts_by_role.get(r, 0) > 0
347 ]
348 open_invitations = (
349 UserInvitation.query.filter_by(tenant_id=tid)
350 .filter(UserInvitation.accepted_at.is_(None))
351 .count()
352 )
353 current_version = os.environ.get("OPENHANGAR_VERSION", "development")
354 latest_setting = db.session.get(AppSetting, "latest_version")
355 latest_version = latest_setting.value if latest_setting else None
356 from utils import check_update_available # pyright: ignore[reportMissingImports]
358 update_available = check_update_available()
359 versions_behind: int | None = None
360 try:
361 import json as _json
363 _all_v_setting = db.session.get(AppSetting, "all_versions")
364 if _all_v_setting and current_version != "development":
365 _all_versions = _json.loads(_all_v_setting.value)
366 if isinstance(_all_versions, list) and current_version in _all_versions:
367 _idx = _all_versions.index(current_version)
368 if _idx > 0:
369 versions_behind = _idx
370 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
371 log.debug("Could not compute versions-behind count: %s", exc)
372 db_size: str | None = None
373 try:
374 from sqlalchemy import text as _text # pyright: ignore[reportMissingImports]
376 _res = db.session.execute(
377 _text("SELECT pg_size_pretty(pg_database_size(current_database()))")
378 ).scalar()
379 db_size = str(_res) if _res is not None else None
380 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
381 log.debug("Could not retrieve DB size: %s", exc)
382 upload_size_bytes: int | None = None
383 try:
384 _upload_folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
385 if os.path.isdir(_upload_folder):
386 upload_size_bytes = sum(
387 os.path.getsize(os.path.join(dp, f))
388 for dp, _dirs, files in os.walk(_upload_folder)
389 for f in files
390 )
391 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
392 log.debug("Could not retrieve upload folder size: %s", exc)
393 backup_total_size_bytes: int | None = None
394 backup_file_count: int = 0
395 try:
396 _backup_folder = current_app.config.get("BACKUP_FOLDER", "/data/backups")
397 if os.path.isdir(_backup_folder):
398 _backup_sizes = [
399 os.path.getsize(os.path.join(dp, f))
400 for dp, _dirs, files in os.walk(_backup_folder)
401 for f in files
402 ]
403 backup_total_size_bytes = sum(_backup_sizes)
404 backup_file_count = len(_backup_sizes)
405 except Exception as exc: # noqa: BLE001 -- best-effort dashboard stat, page must still render
406 log.debug("Could not retrieve backup folder size: %s", exc)
407 from models import Tenant, User # pyright: ignore[reportMissingImports]
409 current_user = db.session.get(User, session["user_id"])
410 tenant_count = Tenant.query.count()
411 _tenant = db.session.get(Tenant, tid) if tid else None
412 upgrade_dir = os.environ.get("OPENHANGAR_UPGRADE_DIR", "")
413 upgrade_dir_enabled = bool(upgrade_dir)
414 upgrade_active = False
415 if upgrade_dir:
416 upgrade_active = os.path.exists(
417 os.path.join(upgrade_dir, "trigger")
418 ) or os.path.exists(os.path.join(upgrade_dir, "trigger.running"))
419 app_debug = current_app.debug
420 sw_forced_on = os.environ.get("OPENHANGAR_SW_ENABLED", "").lower() in (
421 "1",
422 "true",
423 "yes",
424 )
425 sw_server_enabled = not app_debug or sw_forced_on
426 # "demo" is excluded here — _block_in_demo() aborts the whole blueprint
427 # with a 403 before this view runs when OPENHANGAR_ENV=demo.
428 _env_labels = {
429 "production": _("Production"),
430 "development": _("Development"),
431 "test": _("Test"),
432 }
433 _flask_env = os.environ.get("OPENHANGAR_ENV", "production")
434 env_label = _env_labels.get(_flask_env, _flask_env)
435 return render_template(
436 "config/settings.html",
437 records=records,
438 backup_extra=backup_extra,
439 backup_encryption_key_set=bool(_env_or_file("BACKUP_ENCRYPTION_KEY")),
440 backup_folder=current_app.config.get("BACKUP_FOLDER", "/data/backups"),
441 backup_schedule_str=backup_schedule_str,
442 backup_keep=backup_keep,
443 backup_gfs=backup_gfs,
444 last_backup_at=last_backup_at,
445 backup_stale=backup_stale,
446 smtp_status=get_smtp_status(),
447 email_health=get_email_health(),
448 user_counts=user_counts,
449 open_invitations=open_invitations,
450 current_version=current_version,
451 latest_version=latest_version,
452 update_available=update_available,
453 versions_behind=versions_behind,
454 db_size=db_size,
455 upload_size_bytes=upload_size_bytes,
456 backup_total_size_bytes=backup_total_size_bytes,
457 backup_file_count=backup_file_count,
458 current_user=current_user,
459 tenant_count=tenant_count,
460 tenant=_tenant,
461 openaip_api_key=(
462 db.session.get(AppSetting, "openaip_api_key")
463 or type("_", (), {"value": None})()
464 ).value,
465 gatus_configured=_parse_gatus_env() is not None,
466 upgrade_dir_enabled=upgrade_dir_enabled,
467 upgrade_active=upgrade_active,
468 app_debug=app_debug,
469 sw_server_enabled=sw_server_enabled,
470 env_label=env_label,
471 )
474@config_bp.route("/tenant-slug", methods=["POST"])
475@login_required
476def update_tenant_slug() -> ResponseReturnValue:
477 import re as _re
478 import shutil
480 from models import ( # pyright: ignore[reportMissingImports]
481 AircraftPhoto,
482 Document,
483 PendingReconcile,
484 Tenant,
485 TenantUser,
486 )
488 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
489 if not tu:
490 abort(403) # pragma: no cover
491 tenant = db.session.get(Tenant, tu.tenant_id)
492 if not tenant:
493 abort(403) # pragma: no cover
495 raw = request.form.get("slug", "").strip().lower()
496 if not raw:
497 flash(_("Hangar ID cannot be empty."), "danger")
498 return redirect(url_for("config.index"))
500 slug = _re.sub(r"[^a-z0-9]+", "-", raw).strip("-")[:64]
501 if not slug:
502 flash(_("Hangar ID must contain at least one letter or digit."), "danger")
503 return redirect(url_for("config.index"))
505 existing = Tenant.query.filter(Tenant.slug == slug, Tenant.id != tenant.id).first()
506 if existing:
507 flash(_("That Hangar ID is already in use. Please choose another."), "danger")
508 return redirect(url_for("config.index"))
510 old_slug = tenant.slug
511 tenant.slug = slug
513 if old_slug and old_slug != slug:
514 from documents.routes import _safe_join # pyright: ignore[reportMissingImports]
516 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
517 old_dir = _safe_join(folder, old_slug)
518 new_dir = _safe_join(folder, slug)
519 if os.path.isdir(old_dir):
520 if os.path.isdir(new_dir):
521 # Destination already exists — merge file-by-file
522 for dirpath, _dirs, filenames in os.walk(old_dir):
523 rel = os.path.relpath(dirpath, old_dir)
524 dest_dir = os.path.join(new_dir, rel)
525 os.makedirs(dest_dir, exist_ok=True)
526 for fname in filenames:
527 shutil.move(
528 os.path.join(dirpath, fname),
529 os.path.join(dest_dir, fname),
530 )
531 shutil.rmtree(old_dir, ignore_errors=True)
532 else:
533 os.rename(old_dir, new_dir)
535 # Rewrite stored paths in the database
536 prefix_old = old_slug + "/"
537 prefix_new = slug + "/"
538 for doc in Document.query.filter(Document.filename.like(old_slug + "/%")).all():
539 doc.filename = prefix_new + doc.filename[len(prefix_old) :]
540 for pr in PendingReconcile.query.filter(
541 PendingReconcile.filepath.like(old_slug + "/%")
542 ).all():
543 pr.filepath = prefix_new + pr.filepath[len(prefix_old) :]
544 for photo in AircraftPhoto.query.filter(
545 AircraftPhoto.filename.like(old_slug + "/%")
546 ).all():
547 photo.filename = prefix_new + photo.filename[len(prefix_old) :]
549 db.session.commit()
550 flash(_("Hangar ID saved."), "success")
551 return redirect(url_for("config.index"))
554@config_bp.route("/map-tiles", methods=["POST"])
555@login_required
556def update_map_tiles() -> ResponseReturnValue:
557 # ADMIN/OWNER enforcement is handled by config_bp.before_request.
558 key = request.form.get("openaip_api_key", "").strip()
559 setting = db.session.get(AppSetting, "openaip_api_key")
560 if key:
561 if setting:
562 setting.value = key
563 else:
564 db.session.add(AppSetting(key="openaip_api_key", value=key))
565 db.session.commit()
566 flash(_("OpenAIP API key saved."), "success")
567 else:
568 if setting:
569 db.session.delete(setting)
570 db.session.commit()
571 flash(_("OpenAIP API key removed."), "success")
572 return redirect(url_for("config.index"))
575@config_bp.route("/run", methods=["POST"])
576def run_backup_now() -> ResponseReturnValue:
577 if not session.get("user_id"):
578 abort(403)
579 try:
580 record = run_backup()
581 flash(_("Backup completed: %(filename)s", filename=record.filename), "success")
582 except RuntimeError as exc:
583 flash(_("Backup failed: %(error)s", error=exc), "danger")
584 return redirect(url_for("config.index"))
587@config_bp.route("/profile", methods=["POST"])
588def update_profile() -> ResponseReturnValue:
589 if not session.get("user_id"):
590 abort(403)
591 from models import ( # pyright: ignore[reportMissingImports]
592 OperatingModel,
593 Tenant,
594 TenantProfile,
595 TenantUser,
596 )
598 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
599 if not tu:
600 abort(403) # pragma: no cover
602 profile = TenantProfile.query.filter_by(tenant_id=tu.tenant_id).first()
603 if not profile:
604 profile = TenantProfile(tenant_id=tu.tenant_id, setup_complete=True)
605 db.session.add(profile)
607 model_str = request.form.get("operating_model", "sole_operator")
608 try:
609 profile.operating_model = OperatingModel(model_str)
610 except ValueError:
611 flash(_("Invalid operating model."), "danger")
612 return redirect(url_for("config.index"))
613 if model_str == "sole_pilot":
614 profile.planned_aircraft_count = 0
615 profile.allows_rental = False
616 else:
617 try:
618 count = max(1, int(request.form.get("planned_aircraft_count") or 1))
619 except (ValueError, TypeError):
620 count = 1
621 profile.planned_aircraft_count = count
622 profile.allows_rental = bool(request.form.get("allows_rental"))
624 policy = request.form.get("rental_authorization_policy", "").strip()
625 if policy in ("off", "warn", "block"):
626 profile.rental_authorization_policy = policy
628 if profile.operating_model == OperatingModel.SHARED_OWNERSHIP:
629 overdue_raw = request.form.get("co_owner_overdue_days", "").strip()
630 try:
631 overdue_days = int(overdue_raw)
632 if overdue_days >= 1:
633 profile.co_owner_overdue_days = overdue_days
634 except ValueError:
635 pass # non-numeric input — keep the existing setting unchanged
637 tenant = db.session.get(Tenant, tu.tenant_id)
638 if tenant:
639 tenant.require_totp = bool(request.form.get("require_totp"))
641 db.session.commit()
642 flash(_("Usage profile updated."), "success")
643 return redirect(url_for("config.index"))
646@config_bp.route("/email/test", methods=["POST"])
647def test_email() -> ResponseReturnValue:
648 if not session.get("user_id"):
649 abort(403)
650 from models import User # pyright: ignore[reportMissingImports]
651 from services.email_service import (
652 EmailNotConfiguredError,
653 EmailSendError,
654 send_email,
655 ) # pyright: ignore[reportMissingImports]
657 user = db.session.get(User, session["user_id"])
658 if not user:
659 abort(403) # pragma: no cover
660 try:
661 send_email(
662 to=user.email,
663 subject="OpenHangar — test email",
664 text_body=(
665 "This is a test email from your OpenHangar instance.\n\n"
666 "If you received this, your SMTP configuration is working correctly."
667 ),
668 )
669 flash(_("Test email sent to %(email)s.", email=user.email), "success")
670 except EmailNotConfiguredError as exc:
671 flash(_("Email not configured: %(error)s", error=exc), "warning")
672 except EmailSendError as exc:
673 flash(_("Email send failed: %(error)s", error=exc), "danger")
674 return redirect(url_for("config.index"))
677@config_bp.route("/check-version", methods=["POST"])
678def check_version() -> ResponseReturnValue:
679 if not session.get("user_id"):
680 abort(403)
681 import json as _json
682 from datetime import datetime
684 from services.version_service import ( # pyright: ignore[reportMissingImports]
685 fetch_versions,
686 upsert_app_setting,
687 )
689 versions = fetch_versions()
690 upsert_app_setting(
691 db.session,
692 "version_last_checked_at",
693 datetime.now(UTC).isoformat(),
694 )
695 if versions:
696 upsert_app_setting(db.session, "latest_version", versions[0])
697 upsert_app_setting(db.session, "all_versions", _json.dumps(versions))
698 from services.version_service import (
699 _persist_update_flag, # pyright: ignore[reportMissingImports]
700 )
702 _persist_update_flag(
703 db.session, os.environ.get("OPENHANGAR_VERSION", "development"), versions[0]
704 )
705 db.session.commit()
706 flash(_("Version check refreshed."), "success")
707 return redirect(url_for("config.index"))
710# ── One-click upgrade ─────────────────────────────────────────────────────────
713@config_bp.route("/trigger-upgrade", methods=["POST"])
714@require_instance_admin
715def trigger_upgrade() -> ResponseReturnValue:
716 upgrade_dir = os.environ.get("OPENHANGAR_UPGRADE_DIR", "")
717 if not upgrade_dir:
718 abort(404)
719 os.makedirs(upgrade_dir, exist_ok=True)
720 running_path = os.path.join(upgrade_dir, "trigger.running")
721 trigger_path = os.path.join(upgrade_dir, "trigger")
722 if os.path.exists(running_path):
723 flash(_("An upgrade is already in progress."), "warning")
724 return redirect(url_for("config.index"))
725 if os.path.exists(trigger_path):
726 flash(_("Upgrade already triggered."), "info")
727 return redirect(url_for("config.index"))
729 # Guarantee a fallback exists before handing off to the external upgrade
730 # process: once the trigger file is written, it may be picked up at any
731 # time, so the backup must complete first. Block the upgrade if it fails
732 # rather than proceed without a fresh safety net.
733 try:
734 record = run_backup()
735 except RuntimeError as exc:
736 flash(
737 _(
738 "Pre-upgrade backup failed, upgrade not triggered: %(error)s",
739 error=exc,
740 ),
741 "danger",
742 )
743 return redirect(url_for("config.index"))
745 from models import User # pyright: ignore[reportMissingImports]
747 user = db.session.get(User, session["user_id"])
748 trigger_data = {
749 "triggered_by": user.email if user else "unknown",
750 "triggered_at": datetime.now(UTC).isoformat(),
751 }
752 with open(trigger_path, "w") as fh:
753 json.dump(trigger_data, fh)
754 flash(
755 _(
756 "Backup created (%(filename)s). Upgrade triggered — the service "
757 "will restart shortly.",
758 filename=record.filename,
759 ),
760 "info",
761 )
762 return redirect(url_for("config.index"))
765@config_bp.route("/upgrade-status")
766@require_instance_admin
767def upgrade_status() -> ResponseReturnValue:
768 upgrade_dir = os.environ.get("OPENHANGAR_UPGRADE_DIR", "")
769 if not upgrade_dir:
770 return abort(404)
771 done_path = os.path.join(upgrade_dir, "trigger.done")
772 failed_path = os.path.join(upgrade_dir, "trigger.failed")
773 running_path = os.path.join(upgrade_dir, "trigger.running")
774 trigger_path = os.path.join(upgrade_dir, "trigger")
775 if os.path.exists(done_path):
776 with contextlib.suppress(OSError):
777 os.remove(done_path)
778 return jsonify({"status": "done"})
779 if os.path.exists(failed_path):
780 msg = ""
781 with contextlib.suppress(OSError):
782 with open(failed_path) as fh:
783 msg = fh.read().strip()
784 os.remove(failed_path)
785 return jsonify({"status": "failed", "message": msg})
786 if os.path.exists(running_path):
787 return jsonify({"status": "in-progress"})
788 if os.path.exists(trigger_path):
789 return jsonify({"status": "triggered"})
790 return jsonify({"status": "idle"})
793# ── Phase 29: Tenant management (instance admin only) ─────────────────────────
796@config_bp.route("/tenants")
797@require_instance_admin
798def tenant_list() -> ResponseReturnValue:
799 from models import ( # pyright: ignore[reportMissingImports]
800 Aircraft,
801 Role,
802 Tenant,
803 TenantUser,
804 )
806 tenants = Tenant.query.order_by(Tenant.created_at).all()
807 stats = []
808 for t in tenants:
809 user_count = TenantUser.query.filter_by(tenant_id=t.id).count()
810 aircraft_count = Aircraft.query.filter_by(tenant_id=t.id).count()
811 owners = (
812 TenantUser.query.filter_by(tenant_id=t.id)
813 .filter(TenantUser.role.in_([Role.OWNER, Role.ADMIN]))
814 .all()
815 )
816 stats.append(
817 {
818 "tenant": t,
819 "user_count": user_count,
820 "aircraft_count": aircraft_count,
821 "owners": owners,
822 }
823 )
825 return render_template("config/tenant_list.html", stats=stats)
828@config_bp.route("/tenants/create", methods=["GET", "POST"])
829@require_instance_admin
830def tenant_create() -> ResponseReturnValue:
831 from datetime import timedelta
833 from models import ( # pyright: ignore[reportMissingImports]
834 OperatingModel,
835 Role,
836 Tenant,
837 TenantProfile,
838 User,
839 UserInvitation,
840 )
842 user = db.session.get(User, session["user_id"])
843 assert user is not None # guaranteed by @require_instance_admin
845 if request.method == "POST":
846 name = request.form.get("name", "").strip()
847 admin_email = request.form.get("admin_email", "").strip().lower()
848 model_str = request.form.get("operating_model", "sole_operator")
850 if not name:
851 flash(_("Tenant name is required."), "danger")
852 return render_template("config/tenant_create.html")
853 if not admin_email:
854 flash(_("Admin email is required."), "danger")
855 return render_template("config/tenant_create.html")
857 tenant = Tenant(name=name, is_active=True)
858 db.session.add(tenant)
859 db.session.flush()
861 try:
862 op_model: OperatingModel | None = OperatingModel(model_str)
863 except ValueError:
864 op_model = None
866 profile = TenantProfile(
867 tenant_id=tenant.id,
868 operating_model=op_model,
869 setup_complete=False,
870 )
871 db.session.add(profile)
873 invitation = UserInvitation(
874 tenant_id=tenant.id,
875 invited_by_user_id=user.id,
876 email=admin_email,
877 role=Role.OWNER,
878 expires_at=datetime.now(UTC) + timedelta(days=7),
879 )
880 db.session.add(invitation)
881 db.session.commit()
883 flash(
884 _(
885 "Tenant '%(name)s' created. Share this invite link with the owner: %(url)s",
886 name=name,
887 url=url_for(
888 "users.accept_invite", token=invitation.token, _external=True
889 ),
890 ),
891 "success",
892 )
893 return redirect(url_for("config.tenant_list"))
895 return render_template("config/tenant_create.html")
898@config_bp.route("/tenants/<int:tenant_id>/toggle", methods=["POST"])
899@require_instance_admin
900def tenant_toggle_active(tenant_id: int) -> ResponseReturnValue:
901 from models import Tenant # pyright: ignore[reportMissingImports]
903 tenant = db.session.get(Tenant, tenant_id)
904 if not tenant:
905 abort(404)
907 tenant.is_active = not tenant.is_active
908 db.session.commit()
910 if tenant.is_active:
911 flash(_("Tenant '%(name)s' reactivated.", name=tenant.name), "success")
912 else:
913 flash(_("Tenant '%(name)s' deactivated.", name=tenant.name), "warning")
915 return redirect(url_for("config.tenant_list"))
918@config_bp.route("/tenants/<int:tenant_id>/reset-password", methods=["POST"])
919@require_instance_admin
920def tenant_reset_owner_password(tenant_id: int) -> ResponseReturnValue:
921 from datetime import timedelta
923 from models import ( # pyright: ignore[reportMissingImports]
924 PasswordResetToken,
925 Role,
926 Tenant,
927 TenantUser,
928 User,
929 )
931 admin = db.session.get(User, session["user_id"])
932 assert admin is not None # guaranteed by @require_instance_admin
933 tenant = db.session.get(Tenant, tenant_id)
934 if not tenant:
935 abort(404)
937 owner_user_id = request.form.get("owner_user_id", type=int)
938 if not owner_user_id:
939 flash(_("No user selected."), "danger")
940 return redirect(url_for("config.tenant_list"))
942 tu = TenantUser.query.filter_by(tenant_id=tenant_id, user_id=owner_user_id).first()
943 if not tu or tu.role not in (Role.OWNER, Role.ADMIN):
944 abort(403)
946 token = PasswordResetToken(
947 user_id=owner_user_id,
948 generated_by_user_id=admin.id,
949 expires_at=datetime.now(UTC) + timedelta(hours=24),
950 )
951 db.session.add(token)
952 db.session.commit()
954 reset_url = url_for("auth.reset_password", token=token.token, _external=True)
955 return render_template(
956 "config/tenant_reset_token.html",
957 tenant=tenant,
958 reset_url=reset_url,
959 expires_at=token.expires_at,
960 )
963@config_bp.route("/notifications/", methods=["GET", "POST"])
964@login_required
965def notification_preferences() -> ResponseReturnValue:
966 """Manage per-user notification preferences. Accessible to all logged-in users."""
967 from models import ( # pyright: ignore[reportMissingImports]
968 NotificationPreference,
969 NotificationType,
970 Role,
971 TenantNotificationDefault,
972 TenantProfile,
973 TenantUser,
974 User,
975 )
976 from utils import current_user_role # pyright: ignore[reportMissingImports]
978 user = db.session.get(User, session["user_id"])
979 if not user:
980 abort(403)
982 tu = TenantUser.query.filter_by(user_id=user.id).first()
983 tenant_id = tu.tenant_id if tu else None
984 role = current_user_role()
986 is_owner = role in (Role.ADMIN, Role.OWNER)
987 is_pilot = (
988 role in (Role.ADMIN, Role.OWNER, Role.PILOT, Role.INSTRUCTOR) or user.is_pilot
989 )
990 is_maint = (
991 role in (Role.ADMIN, Role.OWNER, Role.MAINTENANCE, Role.INSTRUCTOR)
992 or user.is_maintenance
993 )
995 def _user_has_cap(caps: list[str]) -> bool:
996 return (
997 ("is_owner" in caps and is_owner)
998 or ("is_pilot" in caps and is_pilot)
999 or ("is_maint" in caps and is_maint)
1000 )
1002 visible_types = [
1003 t
1004 for t in NotificationType.ALL
1005 if _user_has_cap(NotificationType.REQUIRED_CAPS.get(t, []))
1006 ]
1008 if request.method == "POST":
1009 if tenant_id is None:
1010 flash(_("Cannot save: no tenant associated."), "danger")
1011 return redirect(url_for("config.notification_preferences"))
1013 for notif_type in visible_types:
1014 enabled = bool(request.form.get(f"enabled_{notif_type}"))
1015 threshold_raw = request.form.get(f"threshold_{notif_type}", "").strip()
1016 threshold_days: int | None = None
1017 if notif_type in NotificationType.HAS_THRESHOLD and threshold_raw:
1018 try:
1019 threshold_days = max(1, int(threshold_raw))
1020 except ValueError:
1021 threshold_days = None
1023 existing = NotificationPreference.query.filter_by(
1024 user_id=user.id, tenant_id=tenant_id, notification_type=notif_type
1025 ).first()
1026 # Only save if the user's preference differs from the effective default
1027 system_default = NotificationType.SYSTEM_DEFAULTS.get(notif_type, {})
1028 same_as_default = enabled == system_default.get(
1029 "enabled", False
1030 ) and threshold_days == system_default.get("threshold_days")
1031 if same_as_default and existing:
1032 db.session.delete(existing)
1033 elif not same_as_default:
1034 if existing:
1035 existing.enabled = enabled
1036 existing.threshold_days = threshold_days
1037 else:
1038 db.session.add(
1039 NotificationPreference(
1040 user_id=user.id,
1041 tenant_id=tenant_id,
1042 notification_type=notif_type,
1043 enabled=enabled,
1044 threshold_days=threshold_days,
1045 )
1046 )
1047 db.session.commit()
1048 flash(_("Notification preferences saved."), "success")
1049 return redirect(url_for("config.notification_preferences"))
1051 # Build current effective preferences for display
1052 prefs: dict[str, dict[str, object]] = {}
1053 for notif_type in visible_types:
1054 if tenant_id:
1055 from services.notification_service import (
1056 get_effective_preference, # pyright: ignore[reportMissingImports]
1057 )
1059 prefs[notif_type] = get_effective_preference(user.id, tenant_id, notif_type)
1060 else:
1061 prefs[notif_type] = dict(
1062 NotificationType.SYSTEM_DEFAULTS.get(
1063 notif_type, {"enabled": False, "threshold_days": None}
1064 )
1065 )
1067 # Tenant defaults visible only to admins/owners
1068 tenant_defaults: dict[str, dict[str, object]] | None = None
1069 if is_owner and tenant_id:
1070 tenant_defaults = {}
1071 for notif_type in NotificationType.ALL:
1072 td = TenantNotificationDefault.query.filter_by(
1073 tenant_id=tenant_id, notification_type=notif_type
1074 ).first()
1075 if td:
1076 tenant_defaults[notif_type] = {
1077 "enabled": td.enabled,
1078 "threshold_days": td.threshold_days,
1079 }
1080 else:
1081 tenant_defaults[notif_type] = dict(
1082 NotificationType.SYSTEM_DEFAULTS.get(
1083 notif_type, {"enabled": False, "threshold_days": None}
1084 )
1085 )
1087 profile = (
1088 TenantProfile.query.filter_by(tenant_id=tenant_id).first()
1089 if tenant_id
1090 else None
1091 )
1093 return render_template(
1094 "config/notifications.html",
1095 visible_types=visible_types,
1096 prefs=prefs,
1097 has_threshold=NotificationType.HAS_THRESHOLD,
1098 system_defaults=NotificationType.SYSTEM_DEFAULTS,
1099 tenant_defaults=tenant_defaults,
1100 is_owner=is_owner,
1101 profile=profile,
1102 )
1105@config_bp.route("/backfill/aircraft-type-icao", methods=["POST"])
1106@require_instance_admin
1107def backfill_aircraft_type_icao() -> ResponseReturnValue:
1108 """Resolve other_aircraft_type_icao for all standalone Flight rows that
1109 have an other_aircraft_type but no icao designator."""
1110 from models import Flight # pyright: ignore[reportMissingImports]
1111 from utils import (
1112 resolve_aircraft_type_icao, # pyright: ignore[reportMissingImports]
1113 )
1115 rows = Flight.query.filter(
1116 Flight.aircraft_id.is_(None),
1117 Flight.other_aircraft_type.isnot(None),
1118 Flight.other_aircraft_type_icao.is_(None),
1119 ).all()
1121 updated = 0
1122 for entry in rows:
1123 resolved = resolve_aircraft_type_icao(entry.other_aircraft_type)
1124 if resolved:
1125 entry.other_aircraft_type_icao = resolved
1126 updated += 1
1128 db.session.commit()
1129 flash(
1130 ngettext(
1131 "Back-fill complete: one of %(total)d entry resolved.",
1132 "Back-fill complete: %(updated)d of %(total)d entries resolved.",
1133 updated,
1134 updated=updated,
1135 total=len(rows),
1136 ),
1137 "success",
1138 )
1139 return redirect(url_for("config.index"))
1142@config_bp.route("/backfill/pilot-log-to-flight-entries", methods=["POST"])
1143@require_instance_admin
1144def backfill_pilot_log_to_flight_entries() -> ResponseReturnValue:
1145 """Promote standalone Flight rows (aircraft_id NULL, other_aircraft_*
1146 registration set) to a managed aircraft, for any whose registration
1147 now matches one."""
1148 from models import Flight # pyright: ignore[reportMissingImports]
1149 from pilots.logbook_import import (
1150 link_entries_to_aircraft, # pyright: ignore[reportMissingImports]
1151 )
1153 entries = (
1154 Flight.query.filter(
1155 Flight.aircraft_id.is_(None),
1156 Flight.other_aircraft_registration.isnot(None),
1157 )
1158 .order_by(Flight.date.asc(), Flight.id.asc())
1159 .all()
1160 )
1162 created = link_entries_to_aircraft(entries)
1163 db.session.commit()
1165 flash(
1166 ngettext(
1167 "Back-fill complete: %(count)d entry linked to a managed aircraft.",
1168 "Back-fill complete: %(count)d entries linked to a managed aircraft.",
1169 created,
1170 count=created,
1171 ),
1172 "success",
1173 )
1174 return redirect(url_for("config.index"))
1177_ALLOWED_BADGE_PATHS: dict[str, str] = {
1178 "uptimes/1h/badge.svg": "uptimes/1h/badge.svg",
1179 "uptimes/24h/badge.svg": "uptimes/24h/badge.svg",
1180 "uptimes/7d/badge.svg": "uptimes/7d/badge.svg",
1181 "uptimes/30d/badge.svg": "uptimes/30d/badge.svg",
1182 "response-times/1h/badge.svg": "response-times/1h/badge.svg",
1183 "response-times/24h/badge.svg": "response-times/24h/badge.svg",
1184 "response-times/7d/badge.svg": "response-times/7d/badge.svg",
1185 "response-times/30d/badge.svg": "response-times/30d/badge.svg",
1186}
1189# ── Renter authorizations (Phase 37c) ───────────────────────────────────────────
1192def _renter_auth_status(auth: Any) -> str:
1193 """'revoked' | 'expired' | 'expiring' | 'valid' — for the list badge."""
1194 from datetime import date as _date
1195 from datetime import timedelta as _timedelta
1197 if auth.revoked_at is not None:
1198 return "revoked"
1199 if not auth.is_valid:
1200 return "expired"
1201 today = _date.today()
1202 soon = today + _timedelta(days=30)
1203 dates = [d for d in (auth.expires_on, auth.medical_valid_until) if d is not None]
1204 if dates and min(dates) <= soon:
1205 return "expiring"
1206 return "valid"
1209def _get_renter_auth_or_404(tenant_id: int, auth_id: int) -> Any:
1210 from models import RenterAuthorization # pyright: ignore[reportMissingImports]
1212 auth = db.session.get(RenterAuthorization, auth_id)
1213 if not auth or auth.tenant_id != tenant_id:
1214 abort(404)
1215 return auth
1218@config_bp.route("/renters/")
1219@login_required
1220def renters_list() -> ResponseReturnValue:
1221 from models import ( # pyright: ignore[reportMissingImports]
1222 RenterAuthorization,
1223 TenantUser,
1224 )
1226 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1227 if not tu:
1228 abort(403) # pragma: no cover
1230 authorizations = (
1231 RenterAuthorization.query.filter_by(tenant_id=tu.tenant_id)
1232 .order_by(RenterAuthorization.granted_on.desc())
1233 .all()
1234 )
1235 rows = [(auth, _renter_auth_status(auth)) for auth in authorizations]
1236 return render_template("config/renters_list.html", rows=rows)
1239def _renter_auth_form_context(tenant_id: int) -> dict[str, Any]:
1240 from models import ( # pyright: ignore[reportMissingImports]
1241 Aircraft,
1242 TenantUser,
1243 User,
1244 )
1246 tenant_users = (
1247 TenantUser.query.filter_by(tenant_id=tenant_id)
1248 .join(User)
1249 .order_by(User.email)
1250 .all()
1251 )
1252 aircraft_list = (
1253 Aircraft.query.filter_by(tenant_id=tenant_id, archived_at=None)
1254 .order_by(Aircraft.registration)
1255 .all()
1256 )
1257 return {"tenant_users": tenant_users, "aircraft_list": aircraft_list}
1260def _save_renter_authorization(tenant_id: int, auth: Any | None) -> ResponseReturnValue:
1261 from datetime import date as _date
1263 from models import ( # pyright: ignore[reportMissingImports]
1264 Aircraft,
1265 RenterAuthorization,
1266 TenantUser,
1267 )
1269 def _date_or_none(key: str) -> _date | None:
1270 raw = request.form.get(key, "").strip()
1271 if not raw:
1272 return None
1273 try:
1274 return _date.fromisoformat(raw)
1275 except ValueError:
1276 return None
1278 renter_user_id_raw = request.form.get("renter_user_id", "").strip()
1279 aircraft_id_raw = request.form.get("aircraft_id", "").strip()
1280 granted_on = _date_or_none("granted_on")
1281 expires_on = _date_or_none("expires_on")
1282 checkout_flight_on = _date_or_none("checkout_flight_on")
1283 licence_seen_on = _date_or_none("licence_seen_on")
1284 medical_valid_until = _date_or_none("medical_valid_until")
1285 notes = request.form.get("notes", "").strip() or None
1287 errors = []
1288 renter_user_id: int | None = None
1289 if renter_user_id_raw:
1290 try:
1291 renter_user_id = int(renter_user_id_raw)
1292 except ValueError:
1293 renter_user_id = None
1294 if (
1295 renter_user_id is None
1296 or not TenantUser.query.filter_by(
1297 tenant_id=tenant_id, user_id=renter_user_id
1298 ).first()
1299 ):
1300 errors.append(_("Select a valid renter."))
1302 aircraft_id: int | None = None
1303 if aircraft_id_raw:
1304 try:
1305 aircraft_id = int(aircraft_id_raw)
1306 except ValueError:
1307 errors.append(_("Invalid aircraft selection."))
1308 else:
1309 if not Aircraft.query.filter_by(
1310 id=aircraft_id, tenant_id=tenant_id
1311 ).first():
1312 errors.append(_("Invalid aircraft selection."))
1314 if granted_on is None:
1315 errors.append(_("Granted date is required and must be a valid date."))
1316 if expires_on is not None and granted_on is not None and expires_on < granted_on:
1317 errors.append(_("Expiry date cannot be before the granted date."))
1319 agreement_file = request.files.get("agreement")
1320 if agreement_file is not None and not agreement_file.filename:
1321 agreement_file = None
1322 if agreement_file is not None:
1323 import os as _os
1325 from documents.routes import (
1326 _ALLOWED_EXTS, # pyright: ignore[reportMissingImports]
1327 )
1329 ext = _os.path.splitext(agreement_file.filename or "")[1].lower()
1330 if ext not in _ALLOWED_EXTS:
1331 errors.append(_("This file type is not allowed for the rental agreement."))
1333 if errors:
1334 for msg in errors:
1335 flash(msg, "danger")
1336 ctx = _renter_auth_form_context(tenant_id)
1337 return render_template("config/renter_form.html", auth=auth, **ctx)
1339 if auth is None:
1340 auth = RenterAuthorization(
1341 tenant_id=tenant_id,
1342 authorized_by_id=session["user_id"],
1343 )
1344 db.session.add(auth)
1346 auth.renter_user_id = renter_user_id
1347 auth.aircraft_id = aircraft_id
1348 auth.granted_on = granted_on
1349 auth.expires_on = expires_on
1350 auth.checkout_flight_on = checkout_flight_on
1351 auth.licence_seen_on = licence_seen_on
1352 auth.medical_valid_until = medical_valid_until
1353 auth.notes = notes
1355 if agreement_file is not None:
1356 from documents.routes import (
1357 _save_upload, # pyright: ignore[reportMissingImports]
1358 )
1359 from models import Document # pyright: ignore[reportMissingImports]
1360 from werkzeug.utils import (
1361 secure_filename, # pyright: ignore[reportMissingImports]
1362 )
1364 db.session.flush()
1365 stored, mime, size = _save_upload(agreement_file, f"renter-agreement-{auth.id}")
1366 db.session.add(
1367 Document(
1368 renter_authorization_id=auth.id,
1369 filename=stored,
1370 original_filename=secure_filename(
1371 agreement_file.filename or "agreement"
1372 ),
1373 mime_type=mime,
1374 size_bytes=size,
1375 title=_("Rental agreement"),
1376 is_sensitive=True,
1377 )
1378 )
1380 db.session.commit()
1381 flash(_("Renter authorization saved."), "success")
1382 return redirect(url_for("config.renters_list"))
1385@config_bp.route("/renters/add", methods=["GET", "POST"])
1386@login_required
1387def renter_add() -> ResponseReturnValue:
1388 from models import TenantUser # pyright: ignore[reportMissingImports]
1390 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1391 if not tu:
1392 abort(403) # pragma: no cover
1394 if request.method == "POST":
1395 return _save_renter_authorization(tu.tenant_id, None)
1397 ctx = _renter_auth_form_context(tu.tenant_id)
1398 return render_template("config/renter_form.html", auth=None, **ctx)
1401@config_bp.route("/renters/<int:auth_id>/edit", methods=["GET", "POST"])
1402@login_required
1403def renter_edit(auth_id: int) -> ResponseReturnValue:
1404 from models import TenantUser # pyright: ignore[reportMissingImports]
1406 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1407 if not tu:
1408 abort(403) # pragma: no cover
1409 auth = _get_renter_auth_or_404(tu.tenant_id, auth_id)
1411 if request.method == "POST":
1412 return _save_renter_authorization(tu.tenant_id, auth)
1414 ctx = _renter_auth_form_context(tu.tenant_id)
1415 return render_template("config/renter_form.html", auth=auth, **ctx)
1418@config_bp.route("/renters/<int:auth_id>/revoke", methods=["POST"])
1419@login_required
1420def renter_revoke(auth_id: int) -> ResponseReturnValue:
1421 from models import TenantUser # pyright: ignore[reportMissingImports]
1423 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1424 if not tu:
1425 abort(403) # pragma: no cover
1426 auth = _get_renter_auth_or_404(tu.tenant_id, auth_id)
1428 auth.revoked_at = datetime.now(UTC)
1429 db.session.commit()
1430 flash(_("Renter authorization revoked."), "success")
1431 return redirect(url_for("config.renters_list"))
1434# ── Renter billing account (Phase 37e) ──────────────────────────────────────────
1437def _renter_account_period(period_months_raw: str | None) -> tuple[Any, Any]:
1438 from datetime import date, timedelta
1440 try:
1441 period_months = int(period_months_raw) if period_months_raw else 12
1442 except ValueError:
1443 period_months = 12
1444 if period_months <= 0:
1445 period_months = 12
1446 end = date.today()
1447 start = end - timedelta(days=period_months * 30)
1448 return start, end
1451@config_bp.route("/renters/<int:user_id>/account")
1452@login_required
1453def renter_account(user_id: int) -> ResponseReturnValue:
1454 from models import ( # pyright: ignore[reportMissingImports]
1455 BillingAccountKind,
1456 TenantUser,
1457 User,
1458 )
1459 from services.billing import BillingService # pyright: ignore[reportMissingImports]
1461 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1462 if not tu:
1463 abort(403) # pragma: no cover
1464 renter_tu = TenantUser.query.filter_by(
1465 user_id=user_id, tenant_id=tu.tenant_id
1466 ).first()
1467 if not renter_tu:
1468 abort(404)
1469 renter = db.session.get(User, user_id)
1470 if renter is None: # pragma: no cover — FK guarantees it exists
1471 abort(404)
1473 account = BillingService.get_or_create_account(
1474 tu.tenant_id, user_id, BillingAccountKind.RENTER
1475 )
1476 db.session.commit() # persist a newly-created account
1477 start, end = _renter_account_period(request.args.get("period"))
1478 statement = BillingService.statement(account, start, end)
1480 return render_template(
1481 "config/renter_account.html",
1482 renter=renter,
1483 account=account,
1484 statement=statement,
1485 balance=BillingService.balance(account),
1486 is_owner_view=True,
1487 csv_url=url_for("config.renter_statement_csv", user_id=user_id),
1488 )
1491@config_bp.route("/renters/<int:user_id>/account/payment", methods=["POST"])
1492@login_required
1493def renter_record_payment(user_id: int) -> ResponseReturnValue:
1494 from models import ( # pyright: ignore[reportMissingImports]
1495 BillingAccountKind,
1496 TenantUser,
1497 )
1498 from services.billing import BillingService # pyright: ignore[reportMissingImports]
1500 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1501 if not tu:
1502 abort(403) # pragma: no cover
1503 renter_tu = TenantUser.query.filter_by(
1504 user_id=user_id, tenant_id=tu.tenant_id
1505 ).first()
1506 if not renter_tu:
1507 abort(404)
1509 from datetime import date as _date_cls
1511 amount_raw = request.form.get("amount", "").strip()
1512 date_raw = request.form.get("date", "").strip()
1513 note = request.form.get("note", "").strip() or None
1515 errors = []
1516 try:
1517 amount = float(amount_raw)
1518 if amount <= 0:
1519 errors.append(_("Payment amount must be positive."))
1520 except ValueError:
1521 amount = 0.0
1522 errors.append(_("Invalid payment amount."))
1523 try:
1524 payment_date = (
1525 _date_cls.fromisoformat(date_raw) if date_raw else _date_cls.today()
1526 )
1527 except ValueError:
1528 payment_date = _date_cls.today()
1529 errors.append(_("Invalid payment date."))
1531 if errors:
1532 for msg in errors:
1533 flash(msg, "danger")
1534 return redirect(url_for("config.renter_account", user_id=user_id))
1536 from models import LedgerEntryType, User # pyright: ignore[reportMissingImports]
1538 account = BillingService.get_or_create_account(
1539 tu.tenant_id, user_id, BillingAccountKind.RENTER
1540 )
1541 recorder = db.session.get(User, session["user_id"])
1542 BillingService.post(
1543 account,
1544 LedgerEntryType.PAYMENT,
1545 -amount,
1546 note or str(_("Payment received")),
1547 payment_date,
1548 source_type="payment",
1549 created_by=recorder,
1550 )
1551 db.session.commit()
1552 flash(_("Payment recorded."), "success")
1553 return redirect(url_for("config.renter_account", user_id=user_id))
1556@config_bp.route("/renters/<int:user_id>/account/statement.csv")
1557@login_required
1558def renter_statement_csv(user_id: int) -> ResponseReturnValue:
1559 from flask import Response # pyright: ignore[reportMissingImports]
1560 from models import ( # pyright: ignore[reportMissingImports]
1561 BillingAccountKind,
1562 TenantUser,
1563 User,
1564 )
1565 from services.billing import BillingService # pyright: ignore[reportMissingImports]
1567 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
1568 if not tu:
1569 abort(403) # pragma: no cover
1570 renter_tu = TenantUser.query.filter_by(
1571 user_id=user_id, tenant_id=tu.tenant_id
1572 ).first()
1573 if not renter_tu:
1574 abort(404)
1576 account = BillingService.get_or_create_account(
1577 tu.tenant_id, user_id, BillingAccountKind.RENTER
1578 )
1579 db.session.commit()
1580 start, end = _renter_account_period(request.args.get("period"))
1581 statement = BillingService.statement(account, start, end)
1582 exporter = db.session.get(User, session["user_id"])
1583 csv_text = BillingService.statement_csv(statement, exported_by=exporter)
1584 return Response(
1585 csv_text,
1586 mimetype="text/csv",
1587 headers={
1588 "Content-Disposition": f"attachment; filename=statement_{user_id}_{start.isoformat()}_{end.isoformat()}.csv"
1589 },
1590 )
1593@config_bp.route("/gatus-badge/<path:badge_path>")
1594@login_required
1595def gatus_badge(badge_path: str) -> ResponseReturnValue:
1596 safe_path = _ALLOWED_BADGE_PATHS.get(badge_path)
1597 if safe_path is None:
1598 return abort(404)
1599 gatus = _parse_gatus_env()
1600 if gatus is None:
1601 return abort(404)
1602 base_url, endpoint_key, auth_header = gatus
1603 badge_url = f"{base_url}/api/v1/endpoints/{endpoint_key}/{safe_path}"
1604 req = urllib.request.Request(badge_url)
1605 if auth_header:
1606 req.add_header("Authorization", f"Basic {auth_header}")
1607 try:
1608 with urllib.request.urlopen(req, timeout=5) as resp: # nosec B310
1609 content = resp.read()
1610 content_type = resp.headers.get("Content-Type", "image/svg+xml")
1611 return current_app.response_class(content, mimetype=content_type)
1612 except urllib.error.URLError as exc:
1613 log.warning("Gatus badge fetch failed (%s): %s", badge_url, repr(exc))
1614 return abort(503)