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

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.helpers import api_publisher_session, api_session 

13from webapp.api.exceptions import ApiResponseError 

14from webapp.extensions import csrf 

15from webapp.login.macaroon import MacaroonRequest, MacaroonResponse 

16from webapp.publisher.snaps import logic 

17 

18login = flask.Blueprint( 

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

20) 

21 

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

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

24 

25 

26# getter for ENVIRONMENT variable 

27# this allows the value to be mocked in tests 

28def get_environment(): 

29 return ENVIRONMENT 

30 

31 

32LP_CANONICAL_TEAM = "canonical" 

33 

34open_id = OpenID( 

35 store_factory=lambda: None, 

36 safe_roots=[], 

37 extension_responses=[MacaroonResponse, TeamsResponse], 

38) 

39 

40dashboard = Dashboard(api_session) 

41publisher_gateway = PublisherGW(api_publisher_session) 

42device_gateway = DeviceGW("snap", api_session) 

43 

44 

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

46@csrf.exempt 

47@open_id.loginhandler 

48def login_handler(): 

49 if authentication.is_authenticated(flask.session): 

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

51 

52 try: 

53 root = authentication.request_macaroon() 

54 except ApiResponseError as api_response_error: 

55 if api_response_error.status_code == 401: 

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

57 else: 

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

59 

60 openid_macaroon = MacaroonRequest( 

61 caveat_id=authentication.get_caveat_id(root) 

62 ) 

63 flask.session["macaroon_root"] = root 

64 

65 lp_teams = TeamsRequest(query_membership=[LP_CANONICAL_TEAM]) 

66 

67 return open_id.try_login( 

68 LOGIN_URL, 

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

70 ask_for_optional=["fullname"], 

71 extensions=[openid_macaroon, lp_teams], 

72 ) 

73 

74 

75@open_id.after_login 

76def after_login(resp): 

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

78 flask.session["macaroon_discharge"] = discharge_macaroon 

79 

80 if not resp.nickname: 

81 return flask.redirect(LOGIN_URL) 

82 

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

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

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

86 flask.session["macaroon_exchanged"] = ( 

87 publisher_gateway.exchange_dashboard_macaroons(flask.session) 

88 ) 

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

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

91 

92 account = dashboard.get_account(flask.session) 

93 validation_sets = dashboard.get_validation_sets(flask.session) 

94 

95 if account: 

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

97 

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

99 # we detect if the user is Canonical by checking 

100 # if the email ends with @canonical.com 

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

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

103 "@canonical.com" 

104 ) 

105 

106 flask.session["publisher"] = { 

107 "identity_url": resp.identity_url, 

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

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

110 "image": resp.image, 

111 "email": account["email"], 

112 "is_canonical": is_canonical, 

113 } 

114 

115 if logic.get_stores( 

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

117 ): 

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

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

120 ) 

121 

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

123 validation_sets is not None 

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

125 ) 

126 else: 

127 flask.session["publisher"] = { 

128 "identity_url": resp.identity_url, 

129 "nickname": resp.nickname, 

130 "fullname": resp.fullname, 

131 "image": resp.image, 

132 "email": resp.email, 

133 } 

134 

135 response = flask.make_response( 

136 flask.redirect( 

137 open_id.get_next_url(), 

138 302, 

139 ), 

140 ) 

141 return response 

142 

143 

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

145def login_beta(): 

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

147 

148 

149@login.route("/logout") 

150def logout(): 

151 authentication.empty_session(flask.session) 

152 

153 return flask.redirect("/")