Coverage for /usr/local/lib/python3.10/site-packages/opal_server-0.0.0-py3.10.egg/opal_server/policy/webhook/deps.py: 26%

1import hashlib 

2import hmac 

3import re 

4from typing import List, Optional 

5 

6from fastapi import Header, HTTPException, Request, status 

7from opal_common.schemas.webhook import GitWebhookRequestParams, SecretTypeEnum 

8from opal_server.config import opal_server_config 

9from pydantic import BaseModel 

10 

11 

12def validate_git_secret_or_throw_factory( 

13 webhook_secret: Optional[str] = opal_server_config.POLICY_REPO_WEBHOOK_SECRET, 

14 webhook_params: GitWebhookRequestParams = opal_server_config.POLICY_REPO_WEBHOOK_PARAMS, 

15): 

16 """Factory function to create secret validator dependency according to 

17 config. 

18 

19 Returns: validate_git_secret_or_throw (async function) 

20 

21 Args: 

22 webhook_secret (Optional[ str ], optional): The secret to validate. Defaults to opal_server_config.POLICY_REPO_WEBHOOK_SECRET. 

23 webhook_params (GitWebhookRequestParams, optional):The webhook configuration - including how to parse the secret. Defaults to opal_server_config.POLICY_REPO_WEBHOOK_PARAMS. 

24 """ 

25 

26 async def validate_git_secret_or_throw(request: Request) -> bool: 

27 """Authenticates a request from a git service webhook system by 

28 checking that the request contains a valid signature (i.e: via the 

29 secret stored on github) or a valid token (as stored in Gitlab).""" 

30 if webhook_secret is None: 30 ↛ 35line 30 didn't jump to line 35 because the condition on line 30 was always true

31 # webhook can be configured without secret (not recommended but quite possible) 

32 return True 

33 

34 # get the secret the git service has sent us 

35 incoming_secret = request.headers.get(webhook_params.secret_header_name, "") 

36 

37 # parse out the actual secret (Some services like Github add prefixes) 

38 matches = re.findall( 

39 webhook_params.secret_parsing_regex, 

40 incoming_secret, 

41 ) 

42 incoming_secret = matches[0] if len(matches) > 0 else None 

43 

44 # check we actually got something 

45 if incoming_secret is None or len(incoming_secret) == 0: 

46 raise HTTPException( 

47 status_code=status.HTTP_401_UNAUTHORIZED, 

48 detail="No secret was provided!", 

49 ) 

50 

51 # Check secret as signature 

52 if webhook_params.secret_type == SecretTypeEnum.signature: 

53 # calculate our signature on the post body 

54 payload = await request.body() 

55 our_signature = hmac.new( 

56 webhook_secret.encode("utf-8"), 

57 payload, 

58 hashlib.sha256, 

59 ).hexdigest() 

60 

61 # compare signatures on the post body 

62 provided_signature = incoming_secret 

63 if not hmac.compare_digest(our_signature, provided_signature): 

64 raise HTTPException( 

65 status_code=status.HTTP_401_UNAUTHORIZED, 

66 detail="signatures didn't match!", 

67 ) 

68 # Check secret as token 

69 elif incoming_secret.encode("utf-8") != webhook_secret.encode("utf-8"): 

70 raise HTTPException( 

71 status_code=status.HTTP_401_UNAUTHORIZED, 

72 detail="secret-tokens didn't match!", 

73 ) 

74 

75 return True 

76 

77 return validate_git_secret_or_throw 

78 

79 

80# Init with defaults 

81validate_git_secret_or_throw = validate_git_secret_or_throw_factory() 

82 

83 

84class GitChanges(BaseModel): 

85 """The summary of a webhook as the properties of what has changed on the 

86 reporting Git repo. 

87 

88 urls - the affected repo URLS 

89 branch - the branch the event affected 

90 """ 

91 

92 urls: List[str] = [] 

93 branch: Optional[str] = None 

94 names: List[str] = [] 

95 

96 

97async def extracted_git_changes(request: Request) -> GitChanges: 

98 """Extracts the repo url from a webhook request payload. 

99 

100 used to make sure that the webhook was triggered on *our* monitored 

101 repo. 

102 

103 This functions search for common patterns for where the affected URL 

104 may appear in the webhook 

105 """ 

106 payload = await request.json() 

107 

108 ### --- Get branch --- ### 

109 # Gitlab / gitHub style 

110 ref = payload.get("ref", None) 

111 

112 # Azure style 

113 if ref is None: 

114 ref = payload.get("refUpdates", {}).get("name", None) 

115 

116 if isinstance(ref, str): 

117 # remove prefix 

118 if ref.startswith("refs/heads/"): 

119 branch = ref[11:] 

120 else: 

121 branch = ref 

122 else: 

123 branch = None 

124 

125 ### Get urls ### 

126 

127 # Github style 

128 repo_payload = payload.get("repository", {}) 

129 git_url = repo_payload.get("git_url", None) 

130 ssh_url = repo_payload.get("ssh_url", None) 

131 clone_url = repo_payload.get("clone_url", None) 

132 

133 # Gitlab style 

134 project_payload = payload.get("project", {}) 

135 project_git_http_url = project_payload.get("git_http_url", None) 

136 project_git_ssh_url = project_payload.get("git_ssh_url", None) 

137 project_full_name = project_payload.get("path_with_namespace", None) 

138 

139 # Azure style 

140 resource_payload = payload.get("resource", {}) 

141 azure_repo_payload = resource_payload.get("repository", {}) 

142 remote_url = azure_repo_payload.get("remoteUrl", None) 

143 

144 # Bitbucket+Github style for fullname 

145 full_name = repo_payload.get("full_name", None) 

146 

147 # additional support for url payload 

148 git_http_url = repo_payload.get("git_ssh_url", None) 

149 ssh_http_url = repo_payload.get("git_http_url", None) 

150 url = repo_payload.get("url", None) 

151 

152 # remove duplicates and None 

153 urls = list( 

154 set( 

155 [ 

156 remote_url, 

157 git_url, 

158 ssh_url, 

159 clone_url, 

160 git_http_url, 

161 ssh_http_url, 

162 url, 

163 project_git_http_url, 

164 project_git_ssh_url, 

165 ] 

166 ) 

167 ) 

168 urls.remove(None) 

169 

170 names = list( 

171 set( 

172 [ 

173 project_full_name, 

174 full_name, 

175 ] 

176 ) 

177 ) 

178 names.remove(None) 

179 

180 if not urls and not names: 

181 raise HTTPException( 

182 status_code=status.HTTP_400_BAD_REQUEST, 

183 detail="repo url or full name not found in payload!", 

184 ) 

185 

186 return GitChanges(urls=urls, branch=branch, names=names)