Coverage for app/pwa/routes.py: 100%
175 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1import contextlib
2import mimetypes as _mimetypes
3import os
4import re as _re
5import shutil
6import tempfile
7import uuid
8from datetime import date as _date
9from typing import cast
11from flask import ( # pyright: ignore[reportMissingImports]
12 Blueprint,
13 abort,
14 current_app,
15 flash,
16 redirect,
17 render_template,
18 request,
19 session,
20 url_for,
21)
22from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
23from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
24from models import ( # pyright: ignore[reportMissingImports]
25 Aircraft,
26 DocCategory,
27 Document,
28 Role,
29 Tenant,
30 TenantUser,
31 db,
32)
33from utils import login_required # pyright: ignore[reportMissingImports]
35pwa_bp = Blueprint("pwa", __name__)
37_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
39# MIME types accepted per destination
40_DEST_ACCEPT: dict[str, frozenset[str]] = {
41 "document": frozenset(
42 {
43 "application/pdf",
44 "image/jpeg",
45 "image/png",
46 "image/gif",
47 "image/webp",
48 "image/heic",
49 "application/msword",
50 "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
51 "application/vnd.ms-excel",
52 "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
53 "text/plain",
54 }
55 ),
56 "expense": frozenset(
57 {"application/pdf", "image/jpeg", "image/png", "image/gif", "image/webp"}
58 ),
59 "maintenance": frozenset({"application/pdf", "image/jpeg", "image/png"}),
60 "flight_photo": frozenset(
61 {"image/jpeg", "image/png", "image/gif", "image/webp", "image/heic"}
62 ),
63}
65# Static MIME-type → file extension mapping. Used to derive the stored
66# extension from the content type rather than the user-supplied filename,
67# which breaks the taint chain for path-injection analysis.
68_MIME_TO_EXT: dict[str, str] = {
69 "application/pdf": ".pdf",
70 "image/jpeg": ".jpg",
71 "image/png": ".png",
72 "image/gif": ".gif",
73 "image/webp": ".webp",
74 "image/heic": ".heic",
75 "application/msword": ".doc",
76 "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
77 "application/vnd.ms-excel": ".xls",
78 "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
79 "text/plain": ".txt",
80}
83def _allowed_destinations(mimetypes: list[str]) -> list[str]:
84 """Return destinations that accept every provided MIME type."""
85 mt_set = set(mimetypes)
86 return [dest for dest in _DEST_ACCEPT if mt_set.issubset(_DEST_ACCEPT[dest])]
89def _dest_labels() -> dict[str, str]:
90 return {
91 "document": _("Aircraft document"),
92 "expense": _("Expense receipt"),
93 "maintenance": _("Maintenance record"),
94 "flight_photo": _("Flight photo"),
95 }
98def _category_labels() -> list[tuple[str, str]]:
99 return [
100 (DocCategory.MAINTENANCE, _("Maintenance")),
101 (DocCategory.INSURANCE, _("Insurance")),
102 (DocCategory.POH, _("POH / Flight Manual")),
103 (DocCategory.AIRWORTHINESS, _("Airworthiness")),
104 (DocCategory.LOGBOOK, _("Logbook")),
105 (DocCategory.INVOICE, _("Invoice")),
106 (DocCategory.OTHER, _("Other")),
107 (DocCategory.UNCATEGORISED, _("Uncategorised")),
108 ]
111def _get_user_aircraft() -> list[Aircraft]:
112 tu = TenantUser.query.filter_by(user_id=session.get("user_id")).first()
113 if not tu:
114 return []
115 return cast(
116 list[Aircraft],
117 Aircraft.query.filter_by(tenant_id=tu.tenant_id)
118 .order_by(Aircraft.registration)
119 .all(),
120 )
123def _ensure_tenant_slug(tenant: Tenant) -> str:
124 if tenant.slug:
125 return str(tenant.slug)
126 base = _re.sub(r"[^a-z0-9]+", "-", tenant.name.lower()).strip("-")[:64]
127 slug = base
128 n = 1
129 while Tenant.query.filter(Tenant.slug == slug, Tenant.id != tenant.id).first():
130 slug = f"{base}-{n}"
131 n += 1
132 tenant.slug = slug
133 db.session.flush()
134 return slug
137def _safe_path_component(s: str) -> str:
138 return _re.sub(r'[<>:"/\\|?*\x00-\x1f]', "", s).strip()
141def _cleanup_temp(tmp_dir: str) -> None:
142 session.pop("share_pending", None)
143 shutil.rmtree(tmp_dir, ignore_errors=True)
146# ── Routes ────────────────────────────────────────────────────────────────────
149@pwa_bp.route("/pwa/shared", methods=["GET"])
150@login_required
151def share_target_get() -> ResponseReturnValue:
152 return redirect(url_for("index"))
155@pwa_bp.route("/pwa/shared", methods=["POST"])
156@login_required
157def share_target() -> ResponseReturnValue:
158 files = request.files.getlist("files")
159 title = request.form.get("title", "").strip()
161 valid_files = [f for f in files if f.filename]
162 if not valid_files:
163 flash(_("No files were shared."), "warning")
164 return redirect(url_for("index"))
166 tmp_dir = tempfile.mkdtemp(prefix="oh-share-")
167 saved: list[dict[str, str]] = []
168 mimetypes: list[str] = []
170 for f in valid_files:
171 original_name = f.filename or "unnamed"
172 safe_name = f"{uuid.uuid4().hex}_{os.path.basename(original_name)}"
173 dest_path = os.path.join(tmp_dir, safe_name)
174 f.save(dest_path)
175 mime = (
176 f.content_type
177 or _mimetypes.guess_type(original_name)[0]
178 or "application/octet-stream"
179 )
180 saved.append({"original": original_name, "saved": safe_name, "mime": mime})
181 mimetypes.append(mime)
183 session["share_pending"] = {
184 "tmp_dir": tmp_dir,
185 "files": saved,
186 "title": title,
187 }
189 destinations = _allowed_destinations(mimetypes)
190 return render_template(
191 "pwa/share_target.html",
192 pending_files=saved,
193 title=title,
194 destinations=destinations,
195 dest_labels=_dest_labels(),
196 aircraft_list=_get_user_aircraft(),
197 categories=_category_labels(),
198 )
201@pwa_bp.route("/pwa/shared/confirm", methods=["POST"])
202@login_required
203def share_confirm() -> ResponseReturnValue:
204 pending = session.get("share_pending")
205 if not pending:
206 flash(_("No pending shared files. Please try sharing again."), "warning")
207 return redirect(url_for("index"))
209 destination = request.form.get("destination", "")
210 tmp_dir: str = pending["tmp_dir"]
211 files_meta: list[dict[str, str]] = pending["files"]
212 title: str = pending.get("title", "")
214 allowed = _allowed_destinations([fm["mime"] for fm in files_meta])
215 if destination not in allowed:
216 _cleanup_temp(tmp_dir)
217 flash(_("Unknown destination."), "danger")
218 return redirect(url_for("index"))
220 if destination == "document":
221 return _process_document(tmp_dir, files_meta, title)
223 if destination == "expense":
224 _cleanup_temp(tmp_dir)
225 flash(_("File received — please attach it manually to the expense."), "info")
226 aircraft_id_raw = request.form.get("aircraft_id", "")
227 if aircraft_id_raw:
228 try:
229 return redirect(
230 url_for("expenses.add_expense", aircraft_id=int(aircraft_id_raw))
231 )
232 except (ValueError, TypeError):
233 pass
234 return redirect(url_for("index"))
236 if destination == "maintenance":
237 _cleanup_temp(tmp_dir)
238 flash(
239 _("File received — please reference it in the maintenance notes."), "info"
240 )
241 aircraft_id_raw = request.form.get("aircraft_id", "")
242 if aircraft_id_raw:
243 try:
244 return redirect(
245 url_for(
246 "maintenance.list_triggers", aircraft_id=int(aircraft_id_raw)
247 )
248 )
249 except (ValueError, TypeError):
250 pass
251 return redirect(url_for("index"))
253 # destination == "flight_photo" — the only remaining option in _DEST_ACCEPT
254 _cleanup_temp(tmp_dir)
255 flash(
256 _("File received — please attach it manually when logging the flight."),
257 "info",
258 )
259 return redirect(url_for("flights.log_flight"))
262def _process_document(
263 tmp_dir: str, files_meta: list[dict[str, str]], title: str
264) -> ResponseReturnValue:
265 tu = TenantUser.query.filter_by(user_id=session.get("user_id")).first()
266 if not tu or tu.role not in _OWNER_ROLES:
267 _cleanup_temp(tmp_dir)
268 abort(403)
270 aircraft_id_raw = request.form.get("aircraft_id", "")
271 try:
272 aircraft_id = int(aircraft_id_raw)
273 except (ValueError, TypeError):
274 _cleanup_temp(tmp_dir)
275 flash(_("Please select an aircraft."), "danger")
276 return redirect(url_for("index"))
278 ac = Aircraft.query.filter_by(id=aircraft_id, tenant_id=tu.tenant_id).first()
279 if not ac:
280 _cleanup_temp(tmp_dir)
281 abort(404)
283 tenant = db.session.get(Tenant, tu.tenant_id)
284 assert tenant is not None # FK guarantees this
286 _raw_category = request.form.get("category") or ""
287 category = next((c for c in DocCategory.ALL if c == _raw_category), None)
288 is_sensitive = bool(request.form.get("is_sensitive"))
289 valid_until_str = request.form.get("valid_until", "").strip()
290 valid_until: _date | None = None
291 if valid_until_str:
292 with contextlib.suppress(ValueError):
293 valid_until = _date.fromisoformat(valid_until_str)
295 doc_title = title.strip() or None
296 upload_folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
298 for file_meta in files_meta:
299 src_path = os.path.join(tmp_dir, file_meta["saved"])
300 original_name = file_meta["original"]
301 mime = file_meta["mime"]
302 ext = next((e for m, e in _MIME_TO_EXT.items() if m == mime), "")
304 if category:
305 slug = _ensure_tenant_slug(tenant)
306 safe_reg = ac.registration.replace("/", "-").replace(" ", "-").upper()
307 today = _date.today().isoformat()
308 safe_t = os.path.basename(
309 _safe_path_component(doc_title or os.path.splitext(original_name)[0])
310 )[:100]
311 fname = f"{today} - {safe_t}{ext}"
312 rel_dir = os.path.join(slug, safe_reg, category)
313 full_dir = os.path.join(upload_folder, rel_dir)
314 os.makedirs(full_dir, exist_ok=True)
315 stored = os.path.join(rel_dir, fname)
316 dest_full = os.path.join(upload_folder, stored)
317 if os.path.exists(dest_full):
318 base, ext2 = os.path.splitext(fname)
319 stored = os.path.join(rel_dir, f"{base}_{uuid.uuid4().hex[:6]}{ext2}")
320 dest_full = os.path.join(upload_folder, stored)
321 else:
322 stored_name = f"doc_share_{uuid.uuid4().hex[:12]}{ext}"
323 os.makedirs(upload_folder, exist_ok=True)
324 stored = stored_name
325 dest_full = os.path.join(upload_folder, stored)
327 shutil.copy2(src_path, dest_full)
328 size = os.path.getsize(dest_full)
330 doc = Document(
331 aircraft_id=ac.id,
332 filename=stored,
333 original_filename=original_name,
334 mime_type=mime,
335 size_bytes=size,
336 title=doc_title,
337 category=category,
338 valid_until=valid_until,
339 is_sensitive=is_sensitive,
340 )
341 db.session.add(doc)
343 db.session.commit()
344 _cleanup_temp(tmp_dir)
345 flash(_("Document uploaded."), "success")
346 return redirect(url_for("documents.list_documents", aircraft_id=ac.id))