Coverage for /usr/local/lib/python3.10/site-packages/opal_server-0.0.0-py3.10.egg/opal_server/policy/watcher/factory.py: 32%

1from functools import partial 

2from typing import Any, List, Optional 

3 

4from fastapi_websocket_pubsub.pub_sub_server import PubSubEndpoint 

5from opal_common.confi.confi import load_conf_if_none 

6from opal_common.git_utils.repo_cloner import RepoClonePathFinder 

7from opal_common.logger import logger 

8from opal_common.sources.api_policy_source import ApiPolicySource 

9from opal_common.sources.git_policy_source import GitPolicySource 

10from opal_common.topics.publisher import TopicPublisher 

11from opal_server.config import PolicySourceTypes, opal_server_config 

12from opal_server.policy.watcher.callbacks import publish_changed_directories 

13from opal_server.policy.watcher.task import BasePolicyWatcherTask, PolicyWatcherTask 

14from opal_server.scopes.task import ScopesPolicyWatcherTask 

15 

16 

17def setup_watcher_task( 

18 publisher: TopicPublisher, 

19 pubsub_endpoint: PubSubEndpoint, 

20 source_type: str = None, 

21 remote_source_url: str = None, 

22 clone_path_finder: RepoClonePathFinder = None, 

23 branch_name: str = None, 

24 ssh_key: Optional[str] = None, 

25 polling_interval: int = None, 

26 request_timeout: int = None, 

27 policy_bundle_token: str = None, 

28 policy_bundle_token_id: str = None, 

29 policy_bundle_server_type: str = None, 

30 policy_bundle_aws_region: str = None, 

31 extensions: Optional[List[str]] = None, 

32 bundle_ignore: Optional[List[str]] = None, 

33) -> BasePolicyWatcherTask: 

34 """Create a PolicyWatcherTask with Git / API policy source defined by env 

35 vars Load all the defaults from config if called without params. 

36 

37 Args: 

38 publisher(TopicPublisher): server side publisher to publish changes in policy 

39 source_type(str): policy source type, can be Git / Api to opa bundle server 

40 remote_source_url(str): the base address to request the policy from 

41 clone_path_finder(RepoClonePathFinder): from which the local dir path for the repo clone would be retrieved 

42 branch_name(str): name of remote branch in git to pull 

43 ssh_key (str, optional): private ssh key used to gain access to the cloned repo 

44 polling_interval(int): how many seconds need to wait between polling 

45 request_timeout(int): how many seconds need to wait until timeout 

46 policy_bundle_token(int): auth token to include in connections to OPAL server. Defaults to POLICY_BUNDLE_SERVER_TOKEN. 

47 policy_bundle_token_id(int): id token to include in connections to OPAL server. Defaults to POLICY_BUNDLE_SERVER_TOKEN_ID. 

48 policy_bundle_server_type (str): type of policy bundle server (HTTP S3). Defaults to POLICY_BUNDLE_SERVER_TYPE 

49 extensions(list(str), optional): list of extantions to check when new policy arrive default is FILTER_FILE_EXTENSIONS 

50 bundle_ignore(list(str), optional): list of glob paths to use for excluding files from bundle default is OPA_BUNDLE_IGNORE 

51 """ 

52 if opal_server_config.SCOPES: 52 ↛ 56line 52 didn't jump to line 56 because the condition on line 52 was always true

53 return ScopesPolicyWatcherTask(pubsub_endpoint) 

54 

55 # load defaults 

56 source_type = load_conf_if_none(source_type, opal_server_config.POLICY_SOURCE_TYPE) 

57 

58 clone_path_finder = load_conf_if_none( 

59 clone_path_finder, 

60 RepoClonePathFinder( 

61 base_clone_path=opal_server_config.POLICY_REPO_CLONE_PATH, 

62 clone_subdirectory_prefix=opal_server_config.POLICY_REPO_CLONE_FOLDER_PREFIX, 

63 use_fixed_path=opal_server_config.POLICY_REPO_REUSE_CLONE_PATH, 

64 ), 

65 ) 

66 

67 clone_path = ( 

68 clone_path_finder.get_clone_path() or clone_path_finder.create_new_clone_path() 

69 ) 

70 logger.info(f"Policy repo will be cloned to: {clone_path}") 

71 

72 branch_name = load_conf_if_none( 

73 branch_name, opal_server_config.POLICY_REPO_MAIN_BRANCH 

74 ) 

75 ssh_key = load_conf_if_none(ssh_key, opal_server_config.POLICY_REPO_SSH_KEY) 

76 polling_interval = load_conf_if_none( 

77 polling_interval, opal_server_config.POLICY_REPO_POLLING_INTERVAL 

78 ) 

79 request_timeout = load_conf_if_none( 

80 request_timeout, opal_server_config.POLICY_REPO_CLONE_TIMEOUT 

81 ) 

82 policy_bundle_token = load_conf_if_none( 

83 policy_bundle_token, opal_server_config.POLICY_BUNDLE_SERVER_TOKEN 

84 ) 

85 extensions = load_conf_if_none( 

86 extensions, opal_server_config.FILTER_FILE_EXTENSIONS 

87 ) 

88 bundle_ignore = load_conf_if_none(bundle_ignore, opal_server_config.BUNDLE_IGNORE) 

89 if source_type == PolicySourceTypes.Git: 

90 remote_source_url = load_conf_if_none( 

91 remote_source_url, opal_server_config.POLICY_REPO_URL 

92 ) 

93 if remote_source_url is None: 

94 logger.warning( 

95 "POLICY_REPO_URL is unset but repo watcher is enabled! disabling watcher." 

96 ) 

97 watcher = GitPolicySource( 

98 remote_source_url=remote_source_url, 

99 local_clone_path=clone_path, 

100 branch_name=branch_name, 

101 ssh_key=ssh_key, 

102 polling_interval=polling_interval, 

103 request_timeout=request_timeout, 

104 ) 

105 elif source_type == PolicySourceTypes.Api: 

106 remote_source_url = load_conf_if_none( 

107 remote_source_url, opal_server_config.POLICY_BUNDLE_URL 

108 ) 

109 if remote_source_url is None: 

110 logger.warning( 

111 "POLICY_BUNDLE_URL is unset but policy watcher is enabled! disabling watcher." 

112 ) 

113 policy_bundle_token_id = load_conf_if_none( 

114 policy_bundle_token_id, opal_server_config.POLICY_BUNDLE_SERVER_TOKEN_ID 

115 ) 

116 policy_bundle_server_type = load_conf_if_none( 

117 policy_bundle_server_type, opal_server_config.POLICY_BUNDLE_SERVER_TYPE 

118 ) 

119 policy_bundle_aws_region = load_conf_if_none( 

120 policy_bundle_aws_region, opal_server_config.POLICY_BUNDLE_SERVER_AWS_REGION 

121 ) 

122 watcher = ApiPolicySource( 

123 remote_source_url=remote_source_url, 

124 local_clone_path=clone_path, 

125 polling_interval=polling_interval, 

126 token=policy_bundle_token, 

127 token_id=policy_bundle_token_id, 

128 bundle_server_type=policy_bundle_server_type, 

129 policy_bundle_path=opal_server_config.POLICY_BUNDLE_TMP_PATH, 

130 policy_bundle_git_add_pattern=opal_server_config.POLICY_BUNDLE_GIT_ADD_PATTERN, 

131 region=policy_bundle_aws_region, 

132 ) 

133 else: 

134 raise ValueError("Unknown value for OPAL_POLICY_SOURCE_TYPE") 

135 watcher.add_on_new_policy_callback( 

136 partial( 

137 publish_changed_directories, 

138 publisher=publisher, 

139 file_extensions=extensions, 

140 bundle_ignore=bundle_ignore, 

141 ) 

142 ) 

143 return PolicyWatcherTask(watcher, pubsub_endpoint)