Coverage for webapp/endpoints/snaps.py: 80%

1import flask 

2from flask import make_response 

3from flask.json import jsonify 

4import json 

5 

6import dns.resolver 

7import re 

8 

9import webapp.helpers as helpers 

10from webapp.decorators import login_required, exchange_required 

11from webapp.store import logic 

12from webapp.config import LP_MAX_BUILD_PAGES, LP_MAX_RECIPES 

13from webapp.api.exceptions import ApiError, ApiTimeoutError 

14from webapp.api.github import repository_is_public 

15from webapp.api.launchpad_provenance import LaunchpadProvenance 

16from webapp.endpoints.utils import get_auditable_map_cache_key 

17from cache.cache_utility import redis_cache 

18 

19from canonicalwebteam.store_api.devicegw import DeviceGW 

20from canonicalwebteam.store_api.dashboard import Dashboard 

21 

22device_gateway = DeviceGW("snap", helpers.api_session) 

23dashboard = Dashboard(helpers.api_session) 

24launchpad_provenance = LaunchpadProvenance() 

25 

26# Fields needed to resolve the default install revision per architecture. 

27AUDITABLE_FIELDS = ["revision", "version", "confinement", "download"] 

28 

29# Bounds the retry rate against Launchpad without pinning a transient 

30# failure for a full hour. 

31FAILED_PROVENANCE_TTL = 60 

32 

33FIELDS = [ 

34 "title", 

35 "summary", 

36 "description", 

37 "license", 

38 "contact", 

39 "website", 

40 "publisher", 

41 "media", 

42 "download", 

43 "version", 

44 "created-at", 

45 "confinement", 

46 "categories", 

47 "trending", 

48 "unlisted", 

49 "links", 

50] 

51snaps = flask.Blueprint( 

52 "snaps", 

53 __name__, 

54) 

55 

56 

57snap_regex = "[a-z0-9-]*[a-z][a-z0-9-]*" 

58 

59 

60def _get_snap_link_fields(snap_name): 

61 try: 

62 details = device_gateway.get_item_details( 

63 snap_name, api_version=2, fields=FIELDS 

64 ) 

65 except Exception: 

66 return None 

67 return { 

68 "links": details["snap"].get("links", {}), 

69 } 

70 

71 

72@snaps.route('/api/<regex("' + snap_regex + '"):snap_name>/verify') 

73def dns_verified_status(snap_name): 

74 res = {"primary_domain": False, "token": None} 

75 context = _get_snap_link_fields(snap_name) 

76 

77 if context is None: 

78 response = make_response(res, 200) 

79 response.cache_control.max_age = FAILED_PROVENANCE_TTL 

80 return response 

81 

82 primary_domain = None 

83 

84 if "website" in context["links"]: 

85 primary_domain = context["links"]["website"][0] 

86 

87 if primary_domain: 

88 token = helpers.get_dns_verification_token(snap_name, primary_domain) 

89 

90 domain = re.compile(r"https?://(www\.)?") 

91 domain = domain.sub("", primary_domain).strip().strip("/") 

92 

93 res["token"] = token 

94 

95 try: 

96 dns_txt_records = [ 

97 dns_record.to_text() 

98 for dns_record in dns.resolver.resolve(domain, "TXT").rrset 

99 ] 

100 

101 if f'"SNAPCRAFT_IO_VERIFICATION={token}"' in dns_txt_records: 

102 res["primary_domain"] = True 

103 

104 except Exception: 

105 res["primary_domain"] = False 

106 

107 response = make_response(res, 200) 

108 response.cache_control.max_age = "3600" 

109 return response 

110 

111 

112def _request_failure_reason(exc): 

113 if isinstance(exc, ApiTimeoutError): 

114 return "store_timeout" 

115 if isinstance(exc, ApiError): 

116 return "store_error" 

117 return "unexpected_error" 

118 

119 

120def _released_revisions(details): 

121 """Every revision currently released in any channel, as strings. 

122 

123 This is what the Security tab lists and always includes the badge's 

124 revision, so a map covering it is complete for both endpoints. 

125 """ 

126 return { 

127 str(channel["revision"]) 

128 for channel in details.get("channel-map") or [] 

129 if channel.get("revision") 

130 } 

131 

132 

133def _get_provenance_map(snap_name, details=None): 

134 """Return the (cached) Launchpad provenance map for a snap. 

135 

136 The map is expensive to build (paginated Launchpad calls), so it is cached 

137 for an hour and shared by both auditable endpoints. 

138 

139 Failures are cached briefly rather than not at all: every page view 

140 fetches provenance, so skipping the cache on failure would send each one 

141 back to an already-struggling Launchpad. The repository check lives here 

142 so its single GitHub call rides the same cache. 

143 """ 

144 cache_key = get_auditable_map_cache_key(snap_name) 

145 cached = redis_cache.get(cache_key, expected_type=dict) 

146 if cached is not None: 

147 return cached 

148 

149 if details is None: 

150 details = device_gateway.get_item_details( 

151 snap_name, api_version=2, fields=AUDITABLE_FIELDS 

152 ) 

153 

154 provenance_map = launchpad_provenance.build_provenance_map( 

155 snap_name, 

156 LP_MAX_BUILD_PAGES, 

157 LP_MAX_RECIPES, 

158 wanted=_released_revisions(details), 

159 ) 

160 provenance_map["source_available"] = repository_is_public( 

161 provenance_map.get("github_repository") 

162 ) 

163 ttl = FAILED_PROVENANCE_TTL if provenance_map.get("failed") else 3600 

164 redis_cache.set(cache_key, provenance_map, ttl=ttl) 

165 return provenance_map 

166 

167 

168def _resolve_default_install(details): 

169 """Resolve the default install option to a single (architecture, revision). 

170 

171 Mirrors what the detail page shows next to the Install button: default 

172 track, lowest available risk, and a deterministic architecture preference 

173 (amd64 if published, otherwise the first architecture sorted). 

174 """ 

175 channel_maps = logic.convert_channel_maps(details.get("channel-map")) 

176 if not channel_maps: 

177 return None, None 

178 

179 default_track = details.get("default-track") or "latest" 

180 lowest_risk = logic.get_lowest_available_risk(channel_maps, default_track) 

181 

182 published = [ 

183 arch 

184 for arch, tracks in channel_maps.items() 

185 if any( 

186 release["risk"] == lowest_risk 

187 for release in tracks.get(default_track, []) 

188 ) 

189 ] 

190 architecture = logic.get_default_architecture( 

191 published or channel_maps.keys() 

192 ) 

193 

194 releases = channel_maps.get(architecture, {}).get(default_track, []) 

195 for release in releases: 

196 if release["risk"] == lowest_risk: 

197 return architecture, release["revision"] 

198 

199 return architecture, None 

200 

201 

202@snaps.route('/api/<regex("' + snap_regex + '"):snap_name>/auditable') 

203def auditable(snap_name): 

204 """Public endpoint backing the provenance badge under the Install button""" 

205 res = {"auditable": False, "status": "not-provided"} 

206 

207 try: 

208 details = device_gateway.get_item_details( 

209 snap_name, api_version=2, fields=AUDITABLE_FIELDS 

210 ) 

211 architecture, revision = _resolve_default_install(details) 

212 

213 if architecture and revision: 

214 provenance_map = _get_provenance_map(snap_name, details) 

215 arch_map = provenance_map.get("revisions", {}).get( 

216 str(revision), {} 

217 ) 

218 build = arch_map.get(architecture) 

219 # The map can span recipes, so prefer the row's own repository. 

220 github_repository = (build or {}).get( 

221 "github_repository" 

222 ) or provenance_map.get("github_repository") 

223 launchpad_repository = (build or {}).get( 

224 "launchpad_repository" 

225 ) or provenance_map.get("launchpad_repository") 

226 

227 base = { 

228 "auditable": False, 

229 "revision": revision, 

230 "architecture": architecture, 

231 } 

232 

233 source_available = provenance_map.get("source_available", True) 

234 

235 if build and build.get("commit_url") and source_available: 

236 res = { 

237 **base, 

238 "auditable": True, 

239 "status": "verified", 

240 "commit_sha": build["commit_sha"], 

241 "github_repository": github_repository, 

242 "launchpad_repository": launchpad_repository, 

243 "commit_url": build["commit_url"], 

244 "build_id": build.get("build_id"), 

245 "build_url": build.get("build_url"), 

246 } 

247 elif build: 

248 # No commit to link (Launchpad-hosted or private source), or 

249 # the repository has gone. Both are settled answers, not 

250 # failed lookups. 

251 res = {**base, "status": "not-provided"} 

252 elif provenance_map.get("failed"): 

253 # Only a real failure earns the error state, since it is the 

254 # only case where "try again later" is true. 

255 res = { 

256 **base, 

257 "status": "error", 

258 "reason": provenance_map.get("reason"), 

259 } 

260 elif github_repository or launchpad_repository: 

261 # Public recipe exists, but this revision has no build/commit. 

262 res = { 

263 **base, 

264 "status": "unavailable", 

265 "github_repository": github_repository, 

266 "launchpad_repository": launchpad_repository, 

267 } 

268 else: 

269 # No public recipe. 

270 res = {**base, "status": "not-provided"} 

271 except Exception as exc: 

272 res = { 

273 "auditable": False, 

274 "status": "error", 

275 "reason": _request_failure_reason(exc), 

276 } 

277 

278 response = make_response(res, 200) 

279 response.cache_control.max_age = ( 

280 FAILED_PROVENANCE_TTL if res.get("status") == "error" else 3600 

281 ) 

282 return response 

283 

284 

285@snaps.route( 

286 '/api/<regex("' + snap_regex + '"):snap_name>/auditable-revisions' 

287) 

288def auditable_revisions(snap_name): 

289 """Public endpoint backing the Security tab's per-revision commit links""" 

290 res = { 

291 "github_repository": None, 

292 "revisions": {}, 

293 "error": False, 

294 "reason": None, 

295 } 

296 

297 try: 

298 provenance_map = _get_provenance_map(snap_name) 

299 res["github_repository"] = provenance_map.get("github_repository") 

300 res["error"] = bool(provenance_map.get("failed")) 

301 res["reason"] = provenance_map.get("reason") 

302 

303 # If the repository is gone, every commit link would 404. 

304 if provenance_map.get("source_available", True): 

305 for revision, arch_map in provenance_map["revisions"].items(): 

306 # One architecture build per store revision. 

307 for build in arch_map.values(): 

308 if build.get("commit_url"): 

309 res["revisions"][revision] = { 

310 "commit_sha": build["commit_sha"], 

311 "commit_url": build["commit_url"], 

312 "build_id": build.get("build_id"), 

313 "build_url": build.get("build_url"), 

314 } 

315 break 

316 except Exception as exc: 

317 res = { 

318 "github_repository": None, 

319 "revisions": {}, 

320 "error": True, 

321 "reason": _request_failure_reason(exc), 

322 } 

323 

324 response = make_response(res, 200) 

325 response.cache_control.max_age = ( 

326 FAILED_PROVENANCE_TTL if res.get("error") else 3600 

327 ) 

328 return response 

329 

330 

331@snaps.route("/api/store/<store_id>/snaps") 

332@login_required 

333@exchange_required 

334def get_store_snaps(store_id): 

335 snaps = dashboard.get_store_snaps(flask.session, store_id) 

336 store = dashboard.get_store(flask.session, store_id) 

337 if "store-whitelist" in store: 

338 included_stores = [] 

339 for item in store["store-whitelist"]: 

340 try: 

341 store_item = dashboard.get_store(flask.session, item) 

342 if store_item: 

343 included_stores.append( 

344 { 

345 "id": store_item["id"], 

346 "name": store_item["name"], 

347 "userHasAccess": True, 

348 } 

349 ) 

350 except Exception: 

351 included_stores.append( 

352 { 

353 "id": item, 

354 "name": "Private store", 

355 "userHasAccess": False, 

356 } 

357 ) 

358 

359 if included_stores: 

360 snaps.append({"included-stores": included_stores}) 

361 return jsonify(snaps) 

362 

363 

364@snaps.route("/api/store/<store_id>/snaps", methods=["POST"]) 

365@login_required 

366@exchange_required 

367def post_manage_store_snaps(store_id): 

368 snaps = json.loads(flask.request.form.get("snaps")) 

369 

370 res = {} 

371 

372 dashboard.update_store_snaps(flask.session, store_id, snaps) 

373 res["msg"] = "Changes saved" 

374 

375 return jsonify({"success": True})