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

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 

11 

12from canonicalwebteam.store_api.devicegw import DeviceGW 

13from canonicalwebteam.store_api.dashboard import Dashboard 

14 

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

16dashboard = Dashboard(helpers.api_session) 

17 

18FIELDS = [ 

19 "title", 

20 "summary", 

21 "description", 

22 "license", 

23 "contact", 

24 "website", 

25 "publisher", 

26 "media", 

27 "download", 

28 "version", 

29 "created-at", 

30 "confinement", 

31 "categories", 

32 "trending", 

33 "unlisted", 

34 "links", 

35] 

36snaps = flask.Blueprint( 

37 "snaps", 

38 __name__, 

39) 

40 

41 

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

43 

44 

45def _get_snap_link_fields(snap_name): 

46 details = device_gateway.get_item_details( 

47 snap_name, api_version=2, fields=FIELDS 

48 ) 

49 context = { 

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

51 } 

52 return context 

53 

54 

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

56def dns_verified_status(snap_name): 

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

58 context = _get_snap_link_fields(snap_name) 

59 

60 primary_domain = None 

61 

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

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

64 

65 if primary_domain: 

66 token = helpers.get_dns_verification_token(snap_name, primary_domain) 

67 

68 domain = re.compile(r"https?://") 

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

70 

71 res["token"] = token 

72 

73 try: 

74 dns_txt_records = [ 

75 dns_record.to_text() 

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

77 ] 

78 

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

80 res["primary_domain"] = True 

81 

82 except Exception: 

83 res["primary_domain"] = False 

84 

85 response = make_response(res, 200) 

86 response.cache_control.max_age = "3600" 

87 return response 

88 

89 

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

91@login_required 

92@exchange_required 

93def get_store_snaps(store_id): 

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

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

96 if "store-whitelist" in store: 

97 included_stores = [] 

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

99 try: 

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

101 if store_item: 

102 included_stores.append( 

103 { 

104 "id": store_item["id"], 

105 "name": store_item["name"], 

106 "userHasAccess": True, 

107 } 

108 ) 

109 except Exception: 

110 included_stores.append( 

111 { 

112 "id": item, 

113 "name": "Private store", 

114 "userHasAccess": False, 

115 } 

116 ) 

117 

118 if included_stores: 

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

120 return jsonify(snaps) 

121 

122 

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

124@login_required 

125@exchange_required 

126def post_manage_store_snaps(store_id): 

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

128 

129 res = {} 

130 

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

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

133 

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