Coverage for webapp / endpoints / signing_keys.py: 96%

1# Packages 

2import flask 

3from flask import make_response 

4from canonicalwebteam.exceptions import ( 

5 StoreApiResponseErrorList, 

6) 

7from canonicalwebteam.store_api.publishergw import PublisherGW 

8 

9# Local 

10from webapp.decorators import login_required, exchange_required 

11from webapp.helpers import api_publisher_session, get_brand_id 

12from webapp.endpoints.models import get_models, get_policies 

13 

14publisher_gateway = PublisherGW("snap", api_publisher_session) 

15 

16signing_keys = flask.Blueprint( 

17 "signing_keys", 

18 __name__, 

19) 

20 

21 

22def get_signing_keys_cache_key(store_id: str) -> str: 

23 return f"{store_id}:signing_keys" 

24 

25 

26@signing_keys.route("/api/store/<store_id>/signing-keys") 

27@login_required 

28@exchange_required 

29def get_signing_keys(store_id: str): 

30 brand_id = get_brand_id(flask.session, store_id) 

31 res = {} 

32 try: 

33 signing_keys = publisher_gateway.get_store_signing_keys( 

34 flask.session, brand_id 

35 ) 

36 res["data"] = signing_keys 

37 res["success"] = True 

38 response = make_response(res, 200) 

39 response.cache_control.max_age = 3600 

40 return response 

41 except StoreApiResponseErrorList as error_list: 

42 res["success"] = False 

43 res["success"] = False 

44 res["message"] = " ".join( 

45 [ 

46 f"{error.get('message', 'An error occurred')}" 

47 for error in error_list.errors 

48 ] 

49 ) 

50 res["data"] = [] 

51 return make_response(res, 500) 

52 

53 

54@signing_keys.route("/api/store/<store_id>/signing-keys", methods=["POST"]) 

55@login_required 

56@exchange_required 

57def create_signing_key(store_id: str): 

58 name = flask.request.form.get("name") 

59 res = {} 

60 

61 try: 

62 if name and len(name) <= 128: 

63 publisher_gateway.create_store_signing_key( 

64 flask.session, store_id, name 

65 ) 

66 res["success"] = True 

67 return make_response(res, 200) 

68 else: 

69 res["message"] = "Invalid signing key. Limit 128 characters" 

70 res["success"] = False 

71 make_response(res, 500) 

72 except StoreApiResponseErrorList as error_list: 

73 res["success"] = False 

74 res["message"] = error_list.errors[0]["message"] 

75 

76 return make_response(res, 500) 

77 

78 

79@signing_keys.route( 

80 "/api/store/<store_id>/signing-keys/<signing_key_sha3_384>", 

81 methods=["DELETE"], 

82) 

83@login_required 

84@exchange_required 

85def delete_signing_key(store_id: str, signing_key_sha3_384: str): 

86 """ 

87 Deletes a signing key from the store. 

88 

89 Args: 

90 store_id (str): The ID of the store. 

91 signing_key_sha3_384 (str): The signing key to delete. 

92 

93 Returns: 

94 Response: A response object with the following fields: 

95 - success (bool): True if the signing key was deleted successfully, 

96 False otherwise. 

97 - message (str): A message describing the result of the deletion. 

98 - data (dict): A dictionary containing models where the signing 

99 key is used. 

100 """ 

101 res = {} 

102 

103 try: 

104 response = publisher_gateway.delete_store_signing_key( 

105 flask.session, store_id, signing_key_sha3_384 

106 ) 

107 

108 if response.status_code == 204: 

109 res["success"] = True 

110 return make_response(res, 200) 

111 elif response.status_code == 404: 

112 res["success"] = False 

113 res["message"] = "Signing key not found" 

114 return make_response(res, 404) 

115 except StoreApiResponseErrorList as error_list: 

116 message = error_list.errors[0]["message"] 

117 if ( 

118 error_list.status_code == 409 

119 and "used to sign at least one serial policy" in message 

120 ): 

121 matching_models = [] 

122 models_response = get_models(store_id).json 

123 models = models_response.get("data", []) 

124 

125 for model in models: 

126 policies_resp = get_policies(store_id, model["name"]).json 

127 policies = policies_resp.get("data", []) 

128 matching_policies = [ 

129 {"revision": policy["revision"]} 

130 for policy in policies 

131 if policy["signing-key-sha3-384"] == signing_key_sha3_384 

132 ] 

133 if matching_policies: 

134 matching_models.append( 

135 { 

136 "name": model["name"], 

137 "policies": matching_policies, 

138 } 

139 ) 

140 res["data"] = {"models": matching_models} 

141 res["message"] = "Signing key is used in at least one policy" 

142 res["success"] = False 

143 else: 

144 res["success"] = False 

145 res["message"] = error_list.errors[0]["message"] 

146 

147 return make_response(res, 500)