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 flask.session["macaroon_discharge"] = resp.extensions["macaroon"].discharge 

78 

79 if not resp.nickname: 

80 return flask.redirect(LOGIN_URL) 

81 

82 account = dashboard.get_account(flask.session) 

83 validation_sets = dashboard.get_validation_sets(flask.session) 

84 

85 if account: 

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

87 

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

89 # we detect if the user is Canonical by checking 

90 # if the email ends with @canonical.com 

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

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

93 "@canonical.com" 

94 ) 

95 

96 flask.session["publisher"] = { 

97 "identity_url": resp.identity_url, 

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

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

100 "image": resp.image, 

101 "email": account["email"], 

102 "is_canonical": is_canonical, 

103 } 

104 

105 if logic.get_stores( 

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

107 ): 

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

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

110 ) 

111 

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

113 validation_sets is not None 

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

115 ) 

116 else: 

117 flask.session["publisher"] = { 

118 "identity_url": resp.identity_url, 

119 "nickname": resp.nickname, 

120 "fullname": resp.fullname, 

121 "image": resp.image, 

122 "email": resp.email, 

123 } 

124 

125 response = flask.make_response( 

126 flask.redirect( 

127 open_id.get_next_url(), 

128 302, 

129 ), 

130 ) 

131 # this is a temporary cookies to be taken out later 

132 response.set_cookie("login_migrated", "true") 

133 return response 

134 

135 

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

137def login_beta(): 

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

139 

140 

141@login.route("/logout") 

142def logout(): 

143 authentication.empty_session(flask.session) 

144 

145 return flask.redirect("/")