1import json 

2import os 

3import hashlib 

4 

5import flask 

6from canonicalwebteam.launchpad import Launchpad 

7from ruamel.yaml import YAML 

8from webapp.api.requests import PublisherSession, Session 

9from canonicalwebteam.store_api.stores.snapstore import SnapPublisher 

10import webapp.api.marketo as marketo_api 

11 

12_yaml = YAML(typ="rt") 

13_yaml_safe = YAML(typ="safe") 

14api_session = Session() 

15api_publisher_session = PublisherSession() 

16marketo = marketo_api.Marketo() 

17publisher_api = SnapPublisher(api_publisher_session) 

18 

19launchpad = Launchpad( 

20 username=os.getenv("LP_API_USERNAME"), 

21 token=os.getenv("LP_API_TOKEN"), 

22 secret=os.getenv("LP_API_TOKEN_SECRET"), 

23 session=api_publisher_session, 

24) 

25 

26 

27def get_yaml_loader(typ="safe"): 

28 if typ == "safe": 

29 return _yaml_safe 

30 return _yaml 

31 

32 

33def get_licenses(): 

34 try: 

35 with open("webapp/licenses.json") as f: 

36 licenses = json.load(f)["licenses"] 

37 

38 def _build_custom_license(license_id, license_name): 

39 return {"licenseId": license_id, "name": license_name} 

40 

41 CUSTOM_LICENSES = [ 

42 _build_custom_license("Proprietary", "Proprietary"), 

43 _build_custom_license("Other Open Source", "Other Open Source"), 

44 _build_custom_license( 

45 "AGPL-3.0+", "GNU Affero General Public License v3.0 or later" 

46 ), 

47 ] 

48 

49 licenses = licenses + CUSTOM_LICENSES 

50 except Exception: 

51 licenses = [] 

52 

53 return licenses 

54 

55 

56def get_file(filename, replaces={}): 

57 """ 

58 Reads a file, replaces occurences of all the keys in `replaces` with 

59 the correspondant values and returns the resulting string or None 

60 

61 Keyword arguments: 

62 filename -- name if the file to load. 

63 replaces -- key/values to replace in the file content (default {}) 

64 """ 

65 filepath = os.path.join(flask.current_app.root_path, filename) 

66 

67 try: 

68 with open(filepath, "r") as f: 

69 data = f.read() 

70 for key in replaces: 

71 data = data.replace(key, replaces[key]) 

72 except Exception: 

73 data = None 

74 

75 return data 

76 

77 

78def get_yaml(filename, typ="safe", replaces={}): 

79 """ 

80 Reads a file, replaces occurences of all the keys in `replaces` with the 

81 correspondant values and returns an ordered dict with the YAML content 

82 

83 Keyword arguments: 

84 filename -- name if the file to load. 

85 typ -- type of yaml loader 

86 replaces -- key/values to replace in the file content (default {}) 

87 """ 

88 try: 

89 yaml = get_yaml_loader(typ) 

90 data = get_file(filename, replaces) 

91 return yaml.load(data) 

92 except Exception: 

93 return None 

94 

95 

96def dump_yaml(data, stream, typ="safe"): 

97 yaml = get_yaml_loader(typ) 

98 yaml.dump(data, stream) 

99 

100 

101def get_icon(media): 

102 icons = [m["url"] for m in media if m["type"] == "icon"] 

103 if len(icons) > 0: 

104 return icons[0] 

105 return "" 

106 

107 

108def get_publisher_data(): 

109 # We don't use the data from this endpoint. 

110 # It is mostly used to make sure the user has signed 

111 # the terms and conditions. 

112 publisher_api.get_account(flask.session) 

113 

114 flask_user = flask.session["publisher"] 

115 

116 subscriptions = None 

117 

118 # don't rely on marketo to show the page, 

119 # if anything fails, just continue and don't show 

120 # this section 

121 try: 

122 subscribed_to_newsletter = False 

123 marketo_user = marketo.get_user(flask_user["email"]) 

124 if marketo_user: 

125 marketo_subscribed = marketo.get_newsletter_subscription( 

126 marketo_user["id"] 

127 ) 

128 if marketo_subscribed.get("snapcraftnewsletter"): 

129 subscribed_to_newsletter = True 

130 

131 subscriptions = {"newsletter": subscribed_to_newsletter} 

132 except Exception: 

133 if "sentry" in flask.current_app.extensions: 

134 flask.current_app.extensions["sentry"].captureException() 

135 

136 flask_user["subscriptions"] = subscriptions 

137 context = {"publisher": flask_user} 

138 

139 return context 

140 

141 

142def get_dns_verification_token(snap_name, domain): 

143 salt = os.getenv("DNS_VERIFICATION_SALT") 

144 token_string = f"{domain}:{snap_name}:{salt}" 

145 token = hashlib.sha256(token_string.encode("utf-8")).hexdigest() 

146 return token 

147 

148 

149def get_csp_as_str(csp={}): 

150 csp_str = "" 

151 for key, values in csp.items(): 

152 csp_value = " ".join(values) 

153 csp_str += f"{key} {csp_value}; " 

154 return csp_str.strip() 

155 

156 

157def list_folders(directory): 

158 return [ 

159 item 

160 for item in os.listdir(directory) 

161 if os.path.isdir(os.path.join(directory, item)) 

162 ]