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

1from typing import Callable, List 

2from urllib.parse import SplitResult, urlparse 

3 

4from fastapi import APIRouter, Depends, Request, status 

5from fastapi_websocket_pubsub.pub_sub_server import PubSubEndpoint 

6from opal_common.authentication.deps import JWTAuthenticator 

7from opal_common.logger import logger 

8from opal_common.schemas.webhook import GitWebhookRequestParams 

9from opal_server.config import PolicySourceTypes, opal_server_config 

10from opal_server.policy.webhook.deps import ( 

11 GitChanges, 

12 extracted_git_changes, 

13 validate_git_secret_or_throw, 

14) 

15 

16 

17def init_git_webhook_router( 

18 pubsub_endpoint: PubSubEndpoint, authenticator: JWTAuthenticator 

19): 

20 async def dummy_affected_repo_urls(request: Request) -> List[str]: 

21 return [] 

22 

23 source_type = opal_server_config.POLICY_SOURCE_TYPE 

24 if source_type == PolicySourceTypes.Api: 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true

25 route_dependency = authenticator 

26 func_dependency = dummy_affected_repo_urls 

27 else: 

28 route_dependency = validate_git_secret_or_throw 

29 func_dependency = extracted_git_changes 

30 

31 return get_webhook_router( 

32 [Depends(route_dependency)], 

33 Depends(func_dependency), 

34 source_type, 

35 pubsub_endpoint.publish, 

36 ) 

37 

38 

39def is_matching_webhook_url(input_url: str, urls: List[str], names: List[str]) -> bool: 

40 parsed = urlparse(input_url) 

41 netloc = parsed.hostname 

42 

43 if parsed.port: 

44 netloc = f"{parsed.hostname}:{parsed.port}" 

45 

46 normalized = SplitResult( 

47 scheme=parsed.scheme, netloc=netloc, path=parsed.path, query="", fragment="" 

48 ) 

49 

50 if urls: 

51 return str(normalized.geturl()) in urls 

52 else: 

53 repo_name_from_path = normalized.path.removeprefix("/").removesuffix(".git") 

54 return repo_name_from_path in names 

55 

56 

57def get_webhook_router( 

58 route_dependencies: List[Depends], 

59 git_changes: Depends, 

60 source_type: PolicySourceTypes, 

61 publish: Callable, 

62 webhook_config: GitWebhookRequestParams = opal_server_config.POLICY_REPO_WEBHOOK_PARAMS, 

63): 

64 if webhook_config is None: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true

65 webhook_config = opal_server_config.POLICY_REPO_WEBHOOK_PARAMS 

66 router = APIRouter() 

67 

68 @router.post( 

69 "/webhook", 

70 status_code=status.HTTP_200_OK, 

71 dependencies=route_dependencies, 

72 ) 

73 async def trigger_webhook(request: Request, git_changes: GitChanges = git_changes): 

74 # TODO: breaking change: change "repo_url" to "remote_url" in next major 

75 if source_type == PolicySourceTypes.Git: 

76 # look at values extracted from request 

77 urls = git_changes.urls 

78 branch = git_changes.branch 

79 names = git_changes.names 

80 

81 # Enforce branch matching (webhook to config) if turned on via config 

82 if ( 

83 opal_server_config.POLICY_REPO_WEBHOOK_ENFORCE_BRANCH 

84 and opal_server_config.POLICY_REPO_MAIN_BRANCH != branch 

85 ): 

86 logger.warning( 

87 "Git Webhook ignored - POLICY_REPO_WEBHOOK_ENFORCE_BRANCH is enabled, and POLICY_REPO_MAIN_BRANCH is `{tracking}` but received webhook for a different branch ({branch})", 

88 tracking=opal_server_config.POLICY_REPO_MAIN_BRANCH, 

89 branch=branch, 

90 ) 

91 return None 

92 

93 # parse event from header 

94 if webhook_config.event_header_name is not None: 

95 event = request.headers.get(webhook_config.event_header_name, "ping") 

96 # parse event from request body 

97 elif webhook_config.event_request_key is not None: 

98 payload = await request.json() 

99 event = payload.get(webhook_config.event_request_key, "ping") 

100 else: 

101 logger.error( 

102 "Webhook config is missing both event_request_key and event_header_name. Must have at least one." 

103 ) 

104 

105 policy_repo_url = opal_server_config.POLICY_REPO_URL 

106 

107 # Check if the URL we are tracking is mentioned in the webhook 

108 if policy_repo_url and ( 

109 is_matching_webhook_url(policy_repo_url, urls, names) 

110 or not webhook_config.match_sender_url 

111 ): 

112 logger.info( 

113 "triggered webhook on repo: {repo}", 

114 repo=opal_server_config.POLICY_REPO_URL, 

115 hook_event=event, 

116 ) 

117 # Check if this it the right event (push) 

118 if event == webhook_config.push_event_value: 

119 # notifies the webhook listener via the pubsub broadcaster 

120 await publish(opal_server_config.POLICY_REPO_WEBHOOK_TOPIC) 

121 return { 

122 "status": "ok", 

123 "event": event, 

124 "repo_url": opal_server_config.POLICY_REPO_URL, 

125 } 

126 else: 

127 logger.warning( 

128 "Got an unexpected webhook not matching the tracked repo ({repo}) - with these URLS: {urls} and those names: {names}.", 

129 repo=opal_server_config.POLICY_REPO_URL, 

130 urls=urls, 

131 names=names, 

132 hook_event=event, 

133 ) 

134 

135 elif source_type == PolicySourceTypes.Api: 

136 logger.info("Triggered webhook to check API bundle URL") 

137 await publish(opal_server_config.POLICY_REPO_WEBHOOK_TOPIC) 

138 return { 

139 "status": "ok", 

140 "event": "webhook_trigger", 

141 "repo_url": opal_server_config.POLICY_BUNDLE_URL, 

142 } 

143 

144 return {"status": "ignored", "event": event} 

145 

146 return router