Coverage for webapp/login/views.py: 88%

1import os 

2 

3import flask 

4from canonicalwebteam.store_api.dashboard import Dashboard 

5from canonicalwebteam.store_api.publishergw import PublisherGW 

6from canonicalwebteam.store_api.devicegw import DeviceGW 

7 

8from django_openid_auth.teams import TeamsRequest, TeamsResponse 

9from flask_openid import OpenID 

10 

11from webapp import authentication 

12from webapp.decorators import login_required 

13from webapp.helpers import api_publisher_session, api_session 

14from webapp.api.exceptions import ApiResponseError 

15from webapp.extensions import csrf 

16from webapp.login.macaroon import MacaroonRequest, MacaroonResponse 

17from webapp.publisher.snaps import logic 

18from webapp.publisher.snaps.build_views import ( 

19 complete_pending_snap_authorization, 

20) 

21from canonicalwebteam.exceptions import StoreApiResponseErrorList 

22 

23login = flask.Blueprint( 

24 "login", __name__, template_folder="/templates", static_folder="/static" 

25) 

26 

27LOGIN_URL = os.getenv("LOGIN_URL", "https://login.ubuntu.com") 

28ENVIRONMENT = os.getenv("ENVIRONMENT", "devel") 

29 

30 

31# getter for ENVIRONMENT variable 

32# this allows the value to be mocked in tests 

33def get_environment(): 

34 return ENVIRONMENT 

35 

36 

37LP_CANONICAL_TEAM = "canonical" 

38 

39open_id = OpenID( 

40 store_factory=lambda: None, 

41 safe_roots=[], 

42 extension_responses=[MacaroonResponse, TeamsResponse], 

43) 

44 

45dashboard = Dashboard(api_session) 

46publisher_gateway = PublisherGW(api_publisher_session) 

47device_gateway = DeviceGW("snap", api_session) 

48 

49 

50@login.route("/login/snap-build-authorization", methods=["GET", "POST"]) 

51@csrf.exempt 

52@open_id.loginhandler 

53@login_required 

54def authorize_snap_build(): 

55 """ 

56 Kick off a discharge round-trip for a snap's Launchpad build/upload 

57 macaroon (see 

58 webapp/publisher/snaps/build_views.py:post_snap_builds). 

59 

60 The macaroon returned by the store's package-upload-macaroon endpoint 

61 carries an SSO third-party caveat that is unique to that macaroon; it 

62 can only be discharged by redirecting the user through login.ubuntu.com 

63 for *that specific* caveat_id. A discharge obtained anywhere else 

64 (e.g. at the user's original login) cannot be reused here. 

65 """ 

66 pending = flask.session.get("pending_snap_authorization") 

67 

68 if not pending: 

69 flask.flash("Nothing to authorize.", "negative") 

70 return flask.redirect("/") 

71 

72 macaroon_request = MacaroonRequest( 

73 caveat_id=authentication.get_caveat_id(pending["root_macaroon"]) 

74 ) 

75 

76 return open_id.try_login( 

77 LOGIN_URL, 

78 extensions=[macaroon_request], 

79 ) 

80 

81 

82@login.route("/login", methods=["GET", "POST"]) 

83@csrf.exempt 

84@open_id.loginhandler 

85def login_handler(): 

86 if authentication.is_authenticated(flask.session): 

87 return flask.redirect(open_id.get_next_url()) 

88 

89 try: 

90 root = authentication.request_macaroon() 

91 except ApiResponseError as api_response_error: 

92 if api_response_error.status_code == 401: 

93 return flask.redirect(flask.url_for(".logout")) 

94 else: 

95 return flask.abort(502, str(api_response_error)) 

96 

97 openid_macaroon = MacaroonRequest( 

98 caveat_id=authentication.get_caveat_id(root) 

99 ) 

100 flask.session["macaroon_root"] = root 

101 

102 lp_teams = TeamsRequest(query_membership=[LP_CANONICAL_TEAM]) 

103 

104 return open_id.try_login( 

105 LOGIN_URL, 

106 ask_for=["email", "nickname", "image"], 

107 ask_for_optional=["fullname"], 

108 extensions=[openid_macaroon, lp_teams], 

109 ) 

110 

111 

112@open_id.after_login 

113def after_login(resp): 

114 # This same OpenID round-trip is reused for two purposes: a normal 

115 # user login, and (see authorize_snap_build above) discharging a 

116 # snap's Launchpad upload macaroon. Handle the latter first and bail 

117 # out early, since none of the account/session logic below applies. 

118 pending = flask.session.pop("pending_snap_authorization", None) 

119 if pending: 

120 discharge_macaroon = resp.extensions["macaroon"].discharge 

121 return complete_pending_snap_authorization(pending, discharge_macaroon) 

122 

123 discharge_macaroon = resp.extensions["macaroon"].discharge 

124 flask.session["macaroon_discharge"] = discharge_macaroon 

125 

126 if not resp.nickname: 

127 return flask.redirect(LOGIN_URL) 

128 

129 # Exchange root + discharge for a single dashboard token. 

130 # Both keys are in the session here, so exchange_dashboard_macaroons 

131 # can read them directly. We then drop them to keep the cookie small. 

132 try: 

133 flask.session["macaroon_exchanged"] = ( 

134 publisher_gateway.exchange_dashboard_macaroons(flask.session) 

135 ) 

136 except StoreApiResponseErrorList as api_error: 

137 # A brand-new publisher who has never accepted the developer Terms & 

138 # Conditions has no publisher account yet, so the macaroon exchange 

139 # fails with "account-not-found". Rather than returning a 404, keep 

140 # the dashboard (root + discharge) macaroons, establish a minimal 

141 # authenticated session and guide the user to the agreement page. 

142 # Accepting the agreement creates the account, after which the 

143 # exchange succeeds on the next request. See issue #5788. 

144 if any( 

145 error.get("code") == "account-not-found" 

146 for error in api_error.errors 

147 ): 

148 flask.session["publisher"] = { 

149 "identity_url": resp.identity_url, 

150 "nickname": resp.nickname, 

151 "fullname": resp.fullname, 

152 "image": resp.image, 

153 "email": resp.email, 

154 } 

155 return flask.redirect(flask.url_for("account.get_agreement")) 

156 raise 

157 

158 flask.session.pop("macaroon_root", None) 

159 flask.session.pop("macaroon_discharge", None) 

160 

161 flask.session["publisher"] = { 

162 "identity_url": resp.identity_url, 

163 "nickname": resp.nickname, 

164 "fullname": resp.fullname, 

165 "image": resp.image, 

166 "email": resp.email, 

167 } 

168 

169 account = dashboard.get_account(flask.session) 

170 validation_sets = dashboard.get_validation_sets(flask.session) 

171 

172 if account: 

173 is_canonical = LP_CANONICAL_TEAM in resp.extensions["lp"].is_member 

174 

175 # in environments other than production, for testing purposes, 

176 # we detect if the user is Canonical by checking 

177 # if the email ends with @canonical.com 

178 if (not is_canonical) and get_environment() != "production": 

179 is_canonical = account["email"] and account["email"].endswith( 

180 "@canonical.com" 

181 ) 

182 

183 flask.session["publisher"] = { 

184 "identity_url": resp.identity_url, 

185 "nickname": account["username"], 

186 "fullname": account["displayname"], 

187 "image": resp.image, 

188 "email": account["email"], 

189 "is_canonical": is_canonical, 

190 } 

191 

192 if logic.get_stores( 

193 account["stores"], roles=["admin", "review", "view"] 

194 ): 

195 flask.session["publisher"]["has_stores"] = ( 

196 len(dashboard.get_stores(flask.session)) > 0 

197 ) 

198 

199 flask.session["publisher"]["has_validation_sets"] = ( 

200 validation_sets is not None 

201 and len(validation_sets["assertions"]) > 0 

202 ) 

203 

204 response = flask.make_response( 

205 flask.redirect( 

206 open_id.get_next_url(), 

207 302, 

208 ), 

209 ) 

210 return response 

211 

212 

213@login.route("/login-beta", methods=["GET"]) 

214def login_beta(): 

215 return flask.redirect(flask.url_for(".login_handler")) 

216 

217 

218@login.route("/logout") 

219def logout(): 

220 authentication.empty_session(flask.session) 

221 

222 return flask.redirect("/")