Coverage for app/share/routes.py: 100%
118 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1"""
2Share blueprint — public read-only aircraft status pages via token.
3"""
5import io
6import secrets
7from datetime import UTC
9from flask import (
10 Blueprint,
11 abort,
12 make_response,
13 redirect,
14 render_template,
15 request,
16 session,
17 url_for,
18) # pyright: ignore[reportMissingImports]
19from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
20from models import (
21 Aircraft,
22 AircraftPhoto,
23 Document,
24 ExpenseType,
25 Flight,
26 MaintenanceTrigger,
27 Role,
28 ShareToken,
29 TenantUser,
30 db,
31) # pyright: ignore[reportMissingImports]
32from utils import login_required, require_role # pyright: ignore[reportMissingImports]
34share_bp = Blueprint("share", __name__)
36_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
38_TOKEN_LENGTH = 16
41def _generate_token() -> str:
42 """Return a unique 16-character URL-safe token (~72 bits of entropy)."""
43 while True:
44 candidate = secrets.token_urlsafe(12)[:_TOKEN_LENGTH]
45 if not ShareToken.query.filter_by(token=candidate).first():
46 return candidate
49def _get_aircraft_or_403(aircraft_id: int) -> Aircraft:
50 """Fetch an aircraft belonging to the logged-in user's tenant, or 403."""
51 from utils import login_required # noqa: F401 — guard already applied by decorator
53 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
54 if not tu:
55 abort(403) # pragma: no cover
56 ac = db.session.get(Aircraft, aircraft_id)
57 if not ac or ac.tenant_id != tu.tenant_id:
58 abort(404)
59 return ac
62# ── Token management (owner-facing) ──────────────────────────────────────────
65@share_bp.route("/aircraft/<aircraft_ref:aircraft_id>/share/create", methods=["POST"])
66@login_required
67@require_role(*_OWNER_ROLES)
68def create_token(aircraft_id: int) -> ResponseReturnValue:
69 ac = _get_aircraft_or_403(aircraft_id)
70 access_level = request.form.get("access_level", "summary")
71 if access_level not in ("summary", "full", "showcase"):
72 access_level = "summary"
73 token = _generate_token()
74 db.session.add(
75 ShareToken(aircraft_id=ac.id, token=token, access_level=access_level)
76 )
77 db.session.commit()
78 return redirect(url_for("aircraft.detail", aircraft_id=aircraft_id))
81@share_bp.route(
82 "/aircraft/<aircraft_ref:aircraft_id>/share/<int:token_id>/revoke", methods=["POST"]
83)
84@login_required
85@require_role(*_OWNER_ROLES)
86def revoke_token(aircraft_id: int, token_id: int) -> ResponseReturnValue:
87 ac = _get_aircraft_or_403(aircraft_id)
88 from datetime import datetime
90 st = db.session.get(ShareToken, token_id)
91 if not st or st.aircraft_id != ac.id:
92 abort(404)
93 st.revoked_at = datetime.now(UTC)
94 db.session.commit()
95 return redirect(url_for("aircraft.detail", aircraft_id=aircraft_id))
98@share_bp.route("/aircraft/<aircraft_ref:aircraft_id>/share/<int:token_id>/qr")
99@login_required
100@require_role(*_OWNER_ROLES)
101def token_qr(aircraft_id: int, token_id: int) -> ResponseReturnValue:
102 ac = _get_aircraft_or_403(aircraft_id)
103 st = db.session.get(ShareToken, token_id)
104 if not st or st.aircraft_id != ac.id or not st.is_active:
105 abort(404)
107 import qrcode # pyright: ignore[reportMissingImports]
109 safe_reg = ac.registration.replace("/", "-").replace(" ", "-")
110 endpoint = (
111 "share.showcase_view" if st.access_level == "showcase" else "share.public_view"
112 )
113 share_url = request.host_url.rstrip("/") + url_for(
114 endpoint, token=st.token, registration=safe_reg
115 )
116 qr = qrcode.QRCode(
117 error_correction=qrcode.constants.ERROR_CORRECT_M, box_size=8, border=4
118 )
119 qr.add_data(share_url)
120 qr.make(fit=True)
121 img = qr.make_image(fill_color="black", back_color="white")
123 buf = io.BytesIO()
124 img.save(buf, format="PNG")
125 buf.seek(0)
127 resp = make_response(buf.read())
128 resp.headers["Content-Type"] = "image/png"
129 resp.headers["Content-Disposition"] = f'attachment; filename="share_{st.token}.png"'
130 return resp
133# ── Public view ───────────────────────────────────────────────────────────────
136@share_bp.route("/share/<token>")
137@share_bp.route("/share/<registration>/<token>")
138def public_view(token: str, registration: str | None = None) -> ResponseReturnValue:
139 # `registration` is a cosmetic prefix only (so an owner can tell at a
140 # glance which aircraft a link is for) — the token remains the sole
141 # lookup/authorization key, so a stale or mismatched registration in
142 # the URL (e.g. after a re-registration) is simply ignored rather than
143 # validated.
144 st = ShareToken.query.filter_by(token=token).first()
145 if not st or not st.is_active or st.access_level == "showcase":
146 abort(404)
148 ac = st.aircraft
149 hobbs = ac.total_engine_hours
150 landings = ac.total_landings
151 flight_hours = ac.total_flight_hours
152 triggers = MaintenanceTrigger.query.filter_by(aircraft_id=ac.id).all()
153 maintenance_summary = [
154 (
155 t,
156 t.status(
157 current_engine_hours=hobbs,
158 current_landings=landings,
159 current_flight_hours=flight_hours,
160 ),
161 )
162 for t in triggers
163 ]
165 overdue = [(t, s) for t, s in maintenance_summary if s == "overdue"]
166 due_soon = [(t, s) for t, s in maintenance_summary if s == "due_soon"]
168 recent_flights = None
169 recent_documents = None
170 if st.access_level == "full":
171 recent_flights = (
172 Flight.query.filter_by(aircraft_id=ac.id)
173 .order_by(Flight.date.desc(), Flight.id.desc())
174 .limit(5)
175 .all()
176 )
177 recent_documents = (
178 Document.query.filter_by(aircraft_id=ac.id, is_sensitive=False)
179 .order_by(Document.uploaded_at.desc())
180 .limit(10)
181 .all()
182 )
184 resp = make_response(
185 render_template(
186 "share/public.html",
187 aircraft=ac,
188 token=st,
189 hobbs=hobbs,
190 flight_hours=flight_hours,
191 maintenance_summary=maintenance_summary,
192 overdue=overdue,
193 due_soon=due_soon,
194 recent_flights=recent_flights,
195 recent_documents=recent_documents,
196 expense_type_labels=ExpenseType.LABELS,
197 )
198 )
199 resp.headers["X-Robots-Tag"] = "noindex, nofollow"
200 return resp
203# ── Public showcase view (backlog: "brag page") ─────────────────────────────
204#
205# A third, lighter share tier, deliberately separate from public_view above:
206# photos and high-level type info only — never flight logs, maintenance
207# status, costs, documents, or pilot names. Looking the token up scoped to
208# access_level == "showcase" (rather than reusing public_view's any-active-
209# token lookup) keeps the tiers from leaking into each other: a showcase
210# token can't be used to pull up the operational public_view page, and a
211# summary/full token can't be used to pull up this one.
214@share_bp.route("/showcase/<token>")
215@share_bp.route("/showcase/<registration>/<token>")
216def showcase_view(token: str, registration: str | None = None) -> ResponseReturnValue:
217 st = ShareToken.query.filter_by(token=token, access_level="showcase").first()
218 if not st or not st.is_active:
219 abort(404)
221 resp = make_response(
222 render_template("share/showcase.html", aircraft=st.aircraft, token=st)
223 )
224 resp.headers["X-Robots-Tag"] = "noindex, nofollow"
225 return resp
228@share_bp.route("/showcase/<token>/photos/<int:photo_id>/img")
229def showcase_photo(token: str, photo_id: int) -> ResponseReturnValue:
230 import os
232 from flask import ( # pyright: ignore[reportMissingImports]
233 current_app,
234 send_from_directory,
235 )
237 st = ShareToken.query.filter_by(token=token, access_level="showcase").first()
238 if not st or not st.is_active:
239 abort(404)
240 photo = db.session.get(AircraftPhoto, photo_id)
241 if not photo or photo.aircraft_id != st.aircraft_id:
242 abort(404)
244 folder = current_app.config.get("UPLOAD_FOLDER", "/data/uploads")
245 directory = os.path.join(folder, os.path.dirname(photo.filename))
246 fname = os.path.basename(photo.filename)
247 response = send_from_directory(directory, fname, max_age=31536000)
248 response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
249 response.headers["X-Robots-Tag"] = "noindex, nofollow"
250 return response