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

1""" 

2Share blueprint — public read-only aircraft status pages via token. 

3""" 

4 

5import io 

6import secrets 

7from datetime import UTC 

8 

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] 

33 

34share_bp = Blueprint("share", __name__) 

35 

36_OWNER_ROLES = (Role.ADMIN, Role.OWNER) 

37 

38_TOKEN_LENGTH = 16 

39 

40 

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 

47 

48 

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 

52 

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 

60 

61 

62# ── Token management (owner-facing) ────────────────────────────────────────── 

63 

64 

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)) 

79 

80 

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 

89 

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)) 

96 

97 

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) 

106 

107 import qrcode # pyright: ignore[reportMissingImports] 

108 

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") 

122 

123 buf = io.BytesIO() 

124 img.save(buf, format="PNG") 

125 buf.seek(0) 

126 

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 

131 

132 

133# ── Public view ─────────────────────────────────────────────────────────────── 

134 

135 

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) 

147 

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 ] 

164 

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"] 

167 

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 ) 

183 

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 

201 

202 

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. 

212 

213 

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) 

220 

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 

226 

227 

228@share_bp.route("/showcase/<token>/photos/<int:photo_id>/img") 

229def showcase_photo(token: str, photo_id: int) -> ResponseReturnValue: 

230 import os 

231 

232 from flask import ( # pyright: ignore[reportMissingImports] 

233 current_app, 

234 send_from_directory, 

235 ) 

236 

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) 

243 

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