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

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.github import repository_is_public 

14from webapp.api.launchpad_provenance import LaunchpadProvenance 

15from webapp.endpoints.utils import get_auditable_map_cache_key 

16from cache.cache_utility import redis_cache 

17 

18from canonicalwebteam.store_api.devicegw import DeviceGW 

19from canonicalwebteam.store_api.dashboard import Dashboard 

20 

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

22dashboard = Dashboard(helpers.api_session) 

23launchpad_provenance = LaunchpadProvenance() 

24 

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

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

27 

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

29# failure for a full hour. 

30FAILED_PROVENANCE_TTL = 60 

31 

32FIELDS = [ 

33 "title", 

34 "summary", 

35 "description", 

36 "license", 

37 "contact", 

38 "website", 

39 "publisher", 

40 "media", 

41 "download", 

42 "version", 

43 "created-at", 

44 "confinement", 

45 "categories", 

46 "trending", 

47 "unlisted", 

48 "links", 

49] 

50snaps = flask.Blueprint( 

51 "snaps", 

52 __name__, 

53) 

54 

55 

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

57 

58 

59def _get_snap_link_fields(snap_name): 

60 details = device_gateway.get_item_details( 

61 snap_name, api_version=2, fields=FIELDS 

62 ) 

63 return { 

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

65 } 

66 

67 

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

69def dns_verified_status(snap_name): 

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

71 context = _get_snap_link_fields(snap_name) 

72 

73 primary_domain = None 

74 

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

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

77 

78 if primary_domain: 

79 token = helpers.get_dns_verification_token(snap_name, primary_domain) 

80 

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

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

83 

84 res["token"] = token 

85 

86 try: 

87 dns_txt_records = [ 

88 dns_record.to_text() 

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

90 ] 

91 

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

93 res["primary_domain"] = True 

94 

95 except Exception: 

96 res["primary_domain"] = False 

97 

98 response = make_response(res, 200) 

99 response.cache_control.max_age = "3600" 

100 return response 

101 

102 

103def _get_provenance_map(snap_name): 

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

105 

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

107 for an hour and shared by both auditable endpoints. 

108 

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

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

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

112 so its single GitHub call rides the same cache. 

113 """ 

114 cache_key = get_auditable_map_cache_key(snap_name) 

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

116 if cached is not None: 

117 return cached 

118 

119 provenance_map = launchpad_provenance.build_provenance_map( 

120 snap_name, LP_MAX_BUILD_PAGES, LP_MAX_RECIPES 

121 ) 

122 provenance_map["source_available"] = repository_is_public( 

123 provenance_map.get("github_repository") 

124 ) 

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

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

127 return provenance_map 

128 

129 

130def _resolve_default_install(details): 

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

132 

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

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

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

136 """ 

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

138 if not channel_maps: 

139 return None, None 

140 

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

142 lowest_risk = logic.get_lowest_available_risk(channel_maps, default_track) 

143 

144 architecture = logic.get_default_architecture(channel_maps.keys()) 

145 

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

147 for release in releases: 

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

149 return architecture, release["revision"] 

150 

151 return architecture, None 

152 

153 

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

155def auditable(snap_name): 

156 """Public endpoint backing the provenance badge under the Install button. 

157 

158 Returns the git commit the default install revision was built from on 

159 Launchpad. The ``status`` field distinguishes every outcome so the badge 

160 can react: 

161 

162 - ``verified``: built on Launchpad from a public commit (commit returned). 

163 - ``unavailable``: a public GitHub recipe exists, but this revision has no 

164 matching build (e.g. uploaded manually). 

165 - ``not-provided``: no public provenance — no Launchpad recipe, or the 

166 recipe's repo is private / non-GitHub. 

167 - ``error``: Launchpad couldn't be reached, so retrying may help. A scan 

168 that merely hit its bounds is normal, not an error. 

169 

170 Never raises to the user. 

171 """ 

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

173 

174 try: 

175 details = device_gateway.get_item_details( 

176 snap_name, api_version=2, fields=AUDITABLE_FIELDS 

177 ) 

178 architecture, revision = _resolve_default_install(details) 

179 

180 if architecture and revision: 

181 provenance_map = _get_provenance_map(snap_name) 

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

183 str(revision), {} 

184 ) 

185 build = arch_map.get(architecture) 

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

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

188 "github_repository" 

189 ) or provenance_map.get("github_repository") 

190 

191 base = { 

192 "auditable": False, 

193 "revision": revision, 

194 "architecture": architecture, 

195 } 

196 

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

198 

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

200 res = { 

201 **base, 

202 "auditable": True, 

203 "status": "verified", 

204 "commit_sha": build["commit_sha"], 

205 "github_repository": github_repository, 

206 "commit_url": build["commit_url"], 

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

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

209 } 

210 elif build: 

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

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

213 # failed lookups. 

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

215 elif provenance_map.get("failed"): 

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

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

218 res = {**base, "status": "error"} 

219 elif github_repository: 

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

221 res = { 

222 **base, 

223 "status": "unavailable", 

224 "github_repository": github_repository, 

225 } 

226 else: 

227 # No public recipe (private or non-GitHub source). 

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

229 except Exception: 

230 res = {"auditable": False, "status": "error"} 

231 

232 response = make_response(res, 200) 

233 response.cache_control.max_age = ( 

234 0 if res.get("status") == "error" else 3600 

235 ) 

236 return response 

237 

238 

239@snaps.route( 

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

241) 

242def auditable_revisions(snap_name): 

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

244 

245 Returns commit links for the snap's recent revisions (bounded by 

246 LP_MAX_BUILD_PAGES). Revisions without provenance are simply absent. 

247 ``error`` is true when Launchpad couldn't be reached, so the Security 

248 tab can distinguish "no provenance" from "couldn't load right now". A 

249 truncated scan is not an error: this endpoint is bounded by design. 

250 """ 

251 res = {"github_repository": None, "revisions": {}, "error": False} 

252 

253 try: 

254 provenance_map = _get_provenance_map(snap_name) 

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

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

257 

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

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

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

261 # One architecture build per store revision. 

262 for build in arch_map.values(): 

263 if build.get("commit_url"): 

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

265 "commit_sha": build["commit_sha"], 

266 "commit_url": build["commit_url"], 

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

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

269 } 

270 break 

271 except Exception: 

272 res = {"github_repository": None, "revisions": {}, "error": True} 

273 

274 response = make_response(res, 200) 

275 response.cache_control.max_age = 0 if res.get("error") else 3600 

276 return response 

277 

278 

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

280@login_required 

281@exchange_required 

282def get_store_snaps(store_id): 

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

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

285 if "store-whitelist" in store: 

286 included_stores = [] 

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

288 try: 

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

290 if store_item: 

291 included_stores.append( 

292 { 

293 "id": store_item["id"], 

294 "name": store_item["name"], 

295 "userHasAccess": True, 

296 } 

297 ) 

298 except Exception: 

299 included_stores.append( 

300 { 

301 "id": item, 

302 "name": "Private store", 

303 "userHasAccess": False, 

304 } 

305 ) 

306 

307 if included_stores: 

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

309 return jsonify(snaps) 

310 

311 

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

313@login_required 

314@exchange_required 

315def post_manage_store_snaps(store_id): 

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

317 

318 res = {} 

319 

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

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

322 

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