Coverage for app/share/routes.py: 100%

1""" 

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

3""" 

4 

5import io 

6import secrets 

7 

8from flask import ( 

9 Blueprint, 

10 abort, 

11 make_response, 

12 redirect, 

13 render_template, 

14 request, 

15 session, 

16 url_for, 

17) # pyright: ignore[reportMissingImports] 

18from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports] 

19 

20from models import ( 

21 Aircraft, 

22 Document, 

23 ExpenseType, 

24 FlightEntry, 

25 MaintenanceTrigger, 

26 Role, 

27 ShareToken, 

28 TenantUser, 

29 db, 

30) # pyright: ignore[reportMissingImports] 

31from utils import login_required, require_role # pyright: ignore[reportMissingImports] 

32 

33share_bp = Blueprint("share", __name__) 

34 

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

36 

37_TOKEN_LENGTH = 16 

38 

39 

40def _generate_token() -> str: 

41 """Return a unique 16-character URL-safe token (~72 bits of entropy).""" 

42 while True: 

43 candidate = secrets.token_urlsafe(12)[:_TOKEN_LENGTH] 

44 if not ShareToken.query.filter_by(token=candidate).first(): 

45 return candidate 

46 

47 

48def _get_aircraft_or_403(aircraft_id: int) -> Aircraft: 

49 """Fetch an aircraft belonging to the logged-in user's tenant, or 403.""" 

50 from utils import login_required # noqa: F401 — guard already applied by decorator 

51 

52 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first() 

53 if not tu: 

54 abort(403) # pragma: no cover 

55 ac = db.session.get(Aircraft, aircraft_id) 

56 if not ac or ac.tenant_id != tu.tenant_id: 

57 abort(404) 

58 return ac 

59 

60 

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

62 

63 

64@share_bp.route("/aircraft/<int:aircraft_id>/share/create", methods=["POST"]) 

65@login_required 

66@require_role(*_OWNER_ROLES) 

67def create_token(aircraft_id: int) -> ResponseReturnValue: 

68 ac = _get_aircraft_or_403(aircraft_id) 

69 access_level = request.form.get("access_level", "summary") 

70 if access_level not in ("summary", "full"): 

71 access_level = "summary" 

72 token = _generate_token() 

73 db.session.add( 

74 ShareToken(aircraft_id=ac.id, token=token, access_level=access_level) 

75 ) 

76 db.session.commit() 

77 return redirect(url_for("aircraft.detail", aircraft_id=aircraft_id)) 

78 

79 

80@share_bp.route( 

81 "/aircraft/<int:aircraft_id>/share/<int:token_id>/revoke", methods=["POST"] 

82) 

83@login_required 

84@require_role(*_OWNER_ROLES) 

85def revoke_token(aircraft_id: int, token_id: int) -> ResponseReturnValue: 

86 ac = _get_aircraft_or_403(aircraft_id) 

87 from datetime import datetime, timezone 

88 

89 st = db.session.get(ShareToken, token_id) 

90 if not st or st.aircraft_id != ac.id: 

91 abort(404) 

92 st.revoked_at = datetime.now(timezone.utc) 

93 db.session.commit() 

94 return redirect(url_for("aircraft.detail", aircraft_id=aircraft_id)) 

95 

96 

97@share_bp.route("/aircraft/<int:aircraft_id>/share/<int:token_id>/qr") 

98@login_required 

99@require_role(*_OWNER_ROLES) 

100def token_qr(aircraft_id: int, token_id: int) -> ResponseReturnValue: 

101 ac = _get_aircraft_or_403(aircraft_id) 

102 st = db.session.get(ShareToken, token_id) 

103 if not st or st.aircraft_id != ac.id or not st.is_active: 

104 abort(404) 

105 

106 import qrcode # pyright: ignore[reportMissingImports] 

107 

108 share_url = request.host_url.rstrip("/") + url_for( 

109 "share.public_view", token=st.token 

110 ) 

111 qr = qrcode.QRCode( 

112 error_correction=qrcode.constants.ERROR_CORRECT_M, box_size=8, border=4 

113 ) 

114 qr.add_data(share_url) 

115 qr.make(fit=True) 

116 img = qr.make_image(fill_color="black", back_color="white") 

117 

118 buf = io.BytesIO() 

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

120 buf.seek(0) 

121 

122 resp = make_response(buf.read()) 

123 resp.headers["Content-Type"] = "image/png" 

124 resp.headers["Content-Disposition"] = f'attachment; filename="share_{st.token}.png"' 

125 return resp 

126 

127 

128# ── Public view ─────────────────────────────────────────────────────────────── 

129 

130 

131@share_bp.route("/share/<token>") 

132def public_view(token: str) -> ResponseReturnValue: 

133 st = ShareToken.query.filter_by(token=token).first() 

134 if not st or not st.is_active: 

135 abort(404) 

136 

137 ac = st.aircraft 

138 hobbs = ac.total_engine_hours 

139 flight_hours = ac.total_flight_hours 

140 triggers = MaintenanceTrigger.query.filter_by(aircraft_id=ac.id).all() 

141 maintenance_summary = [(t, t.status(hobbs)) for t in triggers] 

142 

143 overdue = [(t, s) for t, s in maintenance_summary if s == "overdue"] 

144 due_soon = [(t, s) for t, s in maintenance_summary if s == "due_soon"] 

145 

146 recent_flights = None 

147 recent_documents = None 

148 if st.access_level == "full": 

149 recent_flights = ( 

150 FlightEntry.query.filter_by(aircraft_id=ac.id) 

151 .order_by(FlightEntry.date.desc(), FlightEntry.id.desc()) 

152 .limit(5) 

153 .all() 

154 ) 

155 recent_documents = ( 

156 Document.query.filter_by(aircraft_id=ac.id, is_sensitive=False) 

157 .order_by(Document.uploaded_at.desc()) 

158 .limit(10) 

159 .all() 

160 ) 

161 

162 resp = make_response( 

163 render_template( 

164 "share/public.html", 

165 aircraft=ac, 

166 token=st, 

167 hobbs=hobbs, 

168 flight_hours=flight_hours, 

169 maintenance_summary=maintenance_summary, 

170 overdue=overdue, 

171 due_soon=due_soon, 

172 recent_flights=recent_flights, 

173 recent_documents=recent_documents, 

174 expense_type_labels=ExpenseType.LABELS, 

175 ) 

176 ) 

177 resp.headers["X-Robots-Tag"] = "noindex, nofollow" 

178 return resp