Coverage for webapp/endpoints/snaps.py: 82%
152 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-10 22:08 +0000
1import flask
2from flask import make_response
3from flask.json import jsonify
4import json
6import dns.resolver
7import re
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
19from canonicalwebteam.store_api.devicegw import DeviceGW
20from canonicalwebteam.store_api.dashboard import Dashboard
22device_gateway = DeviceGW("snap", helpers.api_session)
23dashboard = Dashboard(helpers.api_session)
24launchpad_provenance = LaunchpadProvenance()
26# Fields needed to resolve the default install revision per architecture.
27AUDITABLE_FIELDS = ["revision", "version", "confinement", "download"]
29# Bounds the retry rate against Launchpad without pinning a transient
30# failure for a full hour.
31FAILED_PROVENANCE_TTL = 60
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)
57snap_regex = "[a-z0-9-]*[a-z][a-z0-9-]*"
60def _get_snap_link_fields(snap_name):
61 details = device_gateway.get_item_details(
62 snap_name, api_version=2, fields=FIELDS
63 )
64 return {
65 "links": details["snap"].get("links", {}),
66 }
69@snaps.route('/api/<regex("' + snap_regex + '"):snap_name>/verify')
70def dns_verified_status(snap_name):
71 res = {"primary_domain": False, "token": None}
72 context = _get_snap_link_fields(snap_name)
74 primary_domain = None
76 if "website" in context["links"]:
77 primary_domain = context["links"]["website"][0]
79 if primary_domain:
80 token = helpers.get_dns_verification_token(snap_name, primary_domain)
82 domain = re.compile(r"https?://(www\.)?")
83 domain = domain.sub("", primary_domain).strip().strip("/")
85 res["token"] = token
87 try:
88 dns_txt_records = [
89 dns_record.to_text()
90 for dns_record in dns.resolver.resolve(domain, "TXT").rrset
91 ]
93 if f'"SNAPCRAFT_IO_VERIFICATION={token}"' in dns_txt_records:
94 res["primary_domain"] = True
96 except Exception:
97 res["primary_domain"] = False
99 response = make_response(res, 200)
100 response.cache_control.max_age = "3600"
101 return response
104def _request_failure_reason(exc):
105 if isinstance(exc, ApiTimeoutError):
106 return "store_timeout"
107 if isinstance(exc, ApiError):
108 return "store_error"
109 return "unexpected_error"
112def _get_provenance_map(snap_name):
113 """Return the (cached) Launchpad provenance map for a snap.
115 The map is expensive to build (paginated Launchpad calls), so it is cached
116 for an hour and shared by both auditable endpoints.
118 Failures are cached briefly rather than not at all: every page view
119 fetches provenance, so skipping the cache on failure would send each one
120 back to an already-struggling Launchpad. The repository check lives here
121 so its single GitHub call rides the same cache.
122 """
123 cache_key = get_auditable_map_cache_key(snap_name)
124 cached = redis_cache.get(cache_key, expected_type=dict)
125 if cached is not None:
126 return cached
128 provenance_map = launchpad_provenance.build_provenance_map(
129 snap_name, LP_MAX_BUILD_PAGES, LP_MAX_RECIPES
130 )
131 provenance_map["source_available"] = repository_is_public(
132 provenance_map.get("github_repository")
133 )
134 ttl = FAILED_PROVENANCE_TTL if provenance_map.get("failed") else 3600
135 redis_cache.set(cache_key, provenance_map, ttl=ttl)
136 return provenance_map
139def _resolve_default_install(details):
140 """Resolve the default install option to a single (architecture, revision).
142 Mirrors what the detail page shows next to the Install button: default
143 track, lowest available risk, and a deterministic architecture preference
144 (amd64 if published, otherwise the first architecture sorted).
145 """
146 channel_maps = logic.convert_channel_maps(details.get("channel-map"))
147 if not channel_maps:
148 return None, None
150 default_track = details.get("default-track") or "latest"
151 lowest_risk = logic.get_lowest_available_risk(channel_maps, default_track)
153 published = [
154 arch
155 for arch, tracks in channel_maps.items()
156 if any(
157 release["risk"] == lowest_risk
158 for release in tracks.get(default_track, [])
159 )
160 ]
161 architecture = logic.get_default_architecture(
162 published or channel_maps.keys()
163 )
165 releases = channel_maps.get(architecture, {}).get(default_track, [])
166 for release in releases:
167 if release["risk"] == lowest_risk:
168 return architecture, release["revision"]
170 return architecture, None
173@snaps.route('/api/<regex("' + snap_regex + '"):snap_name>/auditable')
174def auditable(snap_name):
175 """Public endpoint backing the provenance badge under the Install button"""
176 res = {"auditable": False, "status": "not-provided"}
178 try:
179 details = device_gateway.get_item_details(
180 snap_name, api_version=2, fields=AUDITABLE_FIELDS
181 )
182 architecture, revision = _resolve_default_install(details)
184 if architecture and revision:
185 provenance_map = _get_provenance_map(snap_name)
186 arch_map = provenance_map.get("revisions", {}).get(
187 str(revision), {}
188 )
189 build = arch_map.get(architecture)
190 # The map can span recipes, so prefer the row's own repository.
191 github_repository = (build or {}).get(
192 "github_repository"
193 ) or provenance_map.get("github_repository")
194 launchpad_repository = (build or {}).get(
195 "launchpad_repository"
196 ) or provenance_map.get("launchpad_repository")
198 base = {
199 "auditable": False,
200 "revision": revision,
201 "architecture": architecture,
202 }
204 source_available = provenance_map.get("source_available", True)
206 if build and build.get("commit_url") and source_available:
207 res = {
208 **base,
209 "auditable": True,
210 "status": "verified",
211 "commit_sha": build["commit_sha"],
212 "github_repository": github_repository,
213 "launchpad_repository": launchpad_repository,
214 "commit_url": build["commit_url"],
215 "build_id": build.get("build_id"),
216 "build_url": build.get("build_url"),
217 }
218 elif build:
219 # No commit to link (Launchpad-hosted or private source), or
220 # the repository has gone. Both are settled answers, not
221 # failed lookups.
222 res = {**base, "status": "not-provided"}
223 elif provenance_map.get("failed"):
224 # Only a real failure earns the error state, since it is the
225 # only case where "try again later" is true.
226 res = {
227 **base,
228 "status": "error",
229 "reason": provenance_map.get("reason"),
230 }
231 elif github_repository or launchpad_repository:
232 # Public recipe exists, but this revision has no build/commit.
233 res = {
234 **base,
235 "status": "unavailable",
236 "github_repository": github_repository,
237 "launchpad_repository": launchpad_repository,
238 }
239 else:
240 # No public recipe.
241 res = {**base, "status": "not-provided"}
242 except Exception as exc:
243 res = {
244 "auditable": False,
245 "status": "error",
246 "reason": _request_failure_reason(exc),
247 }
249 response = make_response(res, 200)
250 response.cache_control.max_age = (
251 FAILED_PROVENANCE_TTL if res.get("status") == "error" else 3600
252 )
253 return response
256@snaps.route(
257 '/api/<regex("' + snap_regex + '"):snap_name>/auditable-revisions'
258)
259def auditable_revisions(snap_name):
260 """Public endpoint backing the Security tab's per-revision commit links"""
261 res = {
262 "github_repository": None,
263 "revisions": {},
264 "error": False,
265 "reason": None,
266 }
268 try:
269 provenance_map = _get_provenance_map(snap_name)
270 res["github_repository"] = provenance_map.get("github_repository")
271 res["error"] = bool(provenance_map.get("failed"))
272 res["reason"] = provenance_map.get("reason")
274 # If the repository is gone, every commit link would 404.
275 if provenance_map.get("source_available", True):
276 for revision, arch_map in provenance_map["revisions"].items():
277 # One architecture build per store revision.
278 for build in arch_map.values():
279 if build.get("commit_url"):
280 res["revisions"][revision] = {
281 "commit_sha": build["commit_sha"],
282 "commit_url": build["commit_url"],
283 "build_id": build.get("build_id"),
284 "build_url": build.get("build_url"),
285 }
286 break
287 except Exception as exc:
288 res = {
289 "github_repository": None,
290 "revisions": {},
291 "error": True,
292 "reason": _request_failure_reason(exc),
293 }
295 response = make_response(res, 200)
296 response.cache_control.max_age = (
297 FAILED_PROVENANCE_TTL if res.get("error") else 3600
298 )
299 return response
302@snaps.route("/api/store/<store_id>/snaps")
303@login_required
304@exchange_required
305def get_store_snaps(store_id):
306 snaps = dashboard.get_store_snaps(flask.session, store_id)
307 store = dashboard.get_store(flask.session, store_id)
308 if "store-whitelist" in store:
309 included_stores = []
310 for item in store["store-whitelist"]:
311 try:
312 store_item = dashboard.get_store(flask.session, item)
313 if store_item:
314 included_stores.append(
315 {
316 "id": store_item["id"],
317 "name": store_item["name"],
318 "userHasAccess": True,
319 }
320 )
321 except Exception:
322 included_stores.append(
323 {
324 "id": item,
325 "name": "Private store",
326 "userHasAccess": False,
327 }
328 )
330 if included_stores:
331 snaps.append({"included-stores": included_stores})
332 return jsonify(snaps)
335@snaps.route("/api/store/<store_id>/snaps", methods=["POST"])
336@login_required
337@exchange_required
338def post_manage_store_snaps(store_id):
339 snaps = json.loads(flask.request.form.get("snaps"))
341 res = {}
343 dashboard.update_store_snaps(flask.session, store_id, snaps)
344 res["msg"] = "Changes saved"
346 return jsonify({"success": True})