Coverage for webapp/decorators.py: 90%

1# Core packages 

2import functools 

3import logging 

4from datetime import datetime, timezone 

5 

6# Third party packages 

7import flask 

8 

9from canonicalwebteam.store_api.dashboard import Dashboard 

10from canonicalwebteam.store_api.publishergw import PublisherGW 

11 

12from webapp import authentication 

13from webapp.helpers import api_publisher_session 

14 

15publisher_gateway = PublisherGW(api_publisher_session) 

16_dashboard = Dashboard(api_publisher_session) 

17logger = logging.getLogger(__name__) 

18 

19# Per-<snap_name> endpoints that must stay reachable even when the snap has 

20# no published revisions. 

21_UNRELEASED_GATE_SKIP_ENDPOINTS = frozenset( 

22 { 

23 "publisher_snaps.delete_package", 

24 "publisher_snaps.get_package_metadata", 

25 "publisher_snaps.get_is_user_snap", 

26 "publisher_snaps.post_github_webhook", 

27 } 

28) 

29 

30 

31def gate_unreleased_snap_pages(): 

32 """ 

33 Block state-changing per-<snap_name> publisher requests when the snap has 

34 no published revisions. Read requests pass through so the page can render 

35 with a warning banner, but saves are rejected to prevent the dashboard API 

36 from returning opaque errors mid-flow. 

37 """ 

38 

39 # Page itself needs to load 

40 if flask.request.method in ("GET", "HEAD", "OPTIONS"): 

41 return None 

42 if not flask.request.view_args: 

43 return None 

44 snap_name = flask.request.view_args.get("snap_name") 

45 if not snap_name: 

46 return None 

47 if flask.request.endpoint in _UNRELEASED_GATE_SKIP_ENDPOINTS: 

48 return None 

49 if not authentication.is_authenticated(flask.session): 

50 return None 

51 

52 try: 

53 history = _dashboard.snap_release_history(flask.session, snap_name, 1) 

54 except Exception: 

55 # If we can't determine release state (dashboard down, network error, 

56 # auth issue), don't block the request. Let the downstream handler 

57 # produce its normal response. 

58 return None 

59 

60 revisions = [] 

61 if isinstance(history, dict): 

62 revisions = history.get("revisions") or [] 

63 elif isinstance(history, list): 

64 revisions = history 

65 

66 if revisions: 

67 return None 

68 

69 return ( 

70 flask.jsonify( 

71 { 

72 "success": False, 

73 "errors": [ 

74 { 

75 "code": "no-releases", 

76 "message": ( 

77 "Publish a first revision before saving " 

78 "changes to this snap." 

79 ), 

80 } 

81 ], 

82 } 

83 ), 

84 403, 

85 ) 

86 

87 

88def login_required(func): 

89 """ 

90 Decorator that checks if a user is logged in, and redirects 

91 to login page if not. 

92 """ 

93 

94 @functools.wraps(func) 

95 def is_user_logged_in(*args, **kwargs): 

96 date = datetime.now(timezone.utc) 

97 date_str = date.strftime("%Y-%m-%dT%H:%M:%S") 

98 

99 if not authentication.is_authenticated(flask.session): 

100 authentication.reset_auth_session(flask.session) 

101 

102 logger.warning( 

103 "User login failed", 

104 extra={ 

105 "datetime": date_str, 

106 "appid": "snapcraft-io", 

107 "event": "authn_login_fail", 

108 }, 

109 ) 

110 

111 return flask.redirect( 

112 flask.url_for("login.login_handler", next=flask.request.path) 

113 ) 

114 

115 publisher = flask.session.get("publisher") 

116 user = publisher["email"] 

117 

118 logger.info( 

119 f"User {user} login successfully", 

120 extra={ 

121 "datetime": date_str, 

122 "appid": "snapcraft-io", 

123 "event": f"authn_login_successafterfail:{user}", 

124 }, 

125 ) 

126 

127 return func(*args, **kwargs) 

128 

129 return is_user_logged_in 

130 

131 

132def exchange_required(func): 

133 @functools.wraps(func) 

134 def is_exchanged(*args, **kwargs): 

135 if "exchanged_developer_token" not in flask.session: 

136 result = publisher_gateway.exchange_dashboard_macaroons( 

137 flask.session 

138 ) 

139 flask.session["developer_token"] = result 

140 flask.session["exchanged_developer_token"] = True 

141 return func(*args, **kwargs) 

142 

143 return is_exchanged