Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/sources/api_policy_source.py: 18%
109 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 11:54 +0000
1from datetime import datetime
2from pathlib import Path
3from typing import Optional, Tuple
4from urllib.parse import urlparse
6import aiohttp
7from fastapi import status
8from fastapi.exceptions import HTTPException
9from opal_common.git_utils.tar_file_to_local_git_extractor import (
10 TarFileToLocalGitExtractor,
11)
12from opal_common.http_utils import redact_url
13from opal_common.logger import logger
14from opal_common.sources.base_policy_source import BasePolicySource
15from opal_common.utils import (
16 build_aws_rest_auth_headers,
17 get_authorization_header,
18 hash_file,
19 throw_if_bad_status_code,
20 tuple_to_dict,
21)
22from opal_server.config import PolicyBundleServerType
23from tenacity import AsyncRetrying
24from tenacity.wait import wait_fixed
26BundleHash = str
29class ApiPolicySource(BasePolicySource):
30 """Watches an OPA-like bundle server for changes and can trigger callbacks
31 when detecting a new bundle.
33 Checking for changes is done by sending an HTTP GET request to the remote bundle server.
34 OPAL will check for changes either when triggered a webhook or periodically if configured
35 to run a polling task.
37 You can read more on OPA bundles here:
38 https://www.openpolicyagent.org/docs/latest/management-bundles/
40 Args:
41 remote_source_url(str): the base address to request the policy from
42 local_clone_path(str): path for the local git to manage policies
43 polling_interval(int): how many seconds need to wait between polling
44 token (str, optional): auth token to include in connections to bundle server. Defaults to POLICY_BUNDLE_SERVER_TOKEN.
45 token_id (str, optional): auth token ID to include in connections to bundle server. Defaults to POLICY_BUNDLE_SERVER_TOKEN_ID.
46 bundle_server_type (PolicyBundleServerType, optional): the type of bundle server
47 """
49 def __init__(
50 self,
51 remote_source_url: str,
52 local_clone_path: str,
53 polling_interval: int = 0,
54 token: Optional[str] = None,
55 token_id: Optional[str] = None,
56 region: Optional[str] = None,
57 bundle_server_type: Optional[PolicyBundleServerType] = None,
58 policy_bundle_path=".",
59 policy_bundle_git_add_pattern="*",
60 ):
61 super().__init__(
62 remote_source_url=remote_source_url,
63 local_clone_path=local_clone_path,
64 polling_interval=polling_interval,
65 )
66 self.token = token
67 self.token_id = token_id
68 self.server_type = bundle_server_type
69 self.region = region
70 self.bundle_hash = None
71 self.etag = None
72 self.tmp_bundle_path = Path(policy_bundle_path)
73 self.policy_bundle_git_add_pattern = policy_bundle_git_add_pattern
74 self.tar_to_git = TarFileToLocalGitExtractor(
75 self.local_clone_path,
76 self.tmp_bundle_path,
77 self.policy_bundle_git_add_pattern,
78 )
80 async def get_initial_policy_state_from_remote(self):
81 """Init remote data to local repo."""
82 async for attempt in AsyncRetrying(wait=wait_fixed(5)):
83 with attempt:
84 try:
85 await self.fetch_policy_bundle_from_api_source(
86 self.remote_source_url, self.token
87 )
88 self.local_git = self.tar_to_git.create_local_git()
89 except Exception:
90 logger.exception(
91 "Failed to load initial policy from remote API bundle server"
92 )
93 raise
95 async def api_update_policy(self) -> Tuple[bool, str, str]:
96 async for attempt in AsyncRetrying(wait=wait_fixed(5)):
97 with attempt:
98 try:
99 (
100 tmp_bundle_path,
101 prev_version,
102 current_hash,
103 ) = await self.fetch_policy_bundle_from_api_source(
104 self.remote_source_url, self.token
105 )
106 if tmp_bundle_path and prev_version and current_hash:
107 commit_msg = f"new version {current_hash}"
108 (
109 self.local_git,
110 prev_commit,
111 new_commit,
112 ) = self.tar_to_git.extract_bundle_to_local_git(
113 commit_msg=commit_msg
114 )
115 return (
116 True,
117 prev_version,
118 current_hash,
119 prev_commit,
120 new_commit,
121 )
122 else:
123 return False, None, current_hash, None, None
124 except Exception as e:
125 logger.exception(
126 f"Failed to update policy from remote API bundle server"
127 )
128 raise
130 def build_auth_headers(self, token=None, path=None):
131 # if it's a simple HTTP server with a bearer token
132 if self.server_type == PolicyBundleServerType.HTTP and token is not None:
133 return tuple_to_dict(get_authorization_header(token))
134 # if it's an AWS s3 server and we have the token and it's id -
135 elif (
136 self.server_type == PolicyBundleServerType.AWS_S3
137 and token is not None
138 and self.token_id is not None
139 ):
140 split_url = urlparse(self.remote_source_url)
141 host = split_url.netloc
142 path = split_url.path + "/" + path
144 return build_aws_rest_auth_headers(
145 self.token_id, token, host, path, self.region
146 )
147 else:
148 return {}
150 async def fetch_policy_bundle_from_api_source(
151 self, url: str, token: Optional[str]
152 ) -> Tuple[Path, BundleHash, BundleHash]:
153 """Fetches the bundle. May throw, in which case we retry again. Checks
154 that the bundle file isn't the same with Etag, if server doesn't have
155 Etag it checks it with hash on the bundle file.
157 Read more on Etag here:
158 https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/ETag
160 Args:
161 url(str): the base address to request the bundle.tar.gz file from
162 token (str, optional): Auth token to include in connections to OPAL server. Defaults to POLICY_BUNDLE_SERVER_TOKEN.
163 Returns:
164 Path: path to the bundle file that we just downloaded from the remote API source
165 BundleHash: previous bundle hash on None if this is the initial bundle file
166 BundleHash: current bundle hash
167 """
168 path = "bundle.tar.gz"
170 auth_headers = self.build_auth_headers(token=token, path=path)
171 etag_headers = (
172 {"ETag": self.etag, "If-None-Match": self.etag} if self.etag else {}
173 )
175 full_url = f"{url}/{path}"
177 async with aiohttp.ClientSession(
178 trust_env=True,
179 ) as session:
180 try:
181 async with session.get(
182 f"{full_url}",
183 headers={
184 "content-type": "application/gzip",
185 **auth_headers,
186 **etag_headers,
187 },
188 ) as response:
189 if response.status == status.HTTP_404_NOT_FOUND:
190 logger.warning(
191 "requested url not found: {full_url}",
192 full_url=redact_url(full_url),
193 )
194 raise HTTPException(
195 status_code=status.HTTP_404_NOT_FOUND,
196 detail=f"requested url not found: {redact_url(full_url)}",
197 )
198 if response.status == status.HTTP_304_NOT_MODIFIED:
199 logger.info(
200 "Not modified at: {now}",
201 now=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
202 )
203 return False, None, self.etag
205 # may throw ValueError
206 await throw_if_bad_status_code(
207 response, expected=[status.HTTP_200_OK], logger=logger
208 )
209 current_etag = response.headers.get("ETag", None)
210 response_bytes = await response.read()
211 tmp_file_path = self.tmp_bundle_path
212 with open(tmp_file_path, "wb") as file:
213 file.write(response_bytes)
215 if not current_etag:
216 logger.info(
217 "Etag is turned off, you may want to turn it on at your bundle server"
218 )
219 current_bundle_hash = hash_file(tmp_file_path)
220 logger.info("Bundle hash is {hash}", hash=current_bundle_hash)
221 if self.bundle_hash == current_bundle_hash:
222 logger.info(
223 "No new bundle, hash is: {hash}",
224 hash=current_bundle_hash,
225 )
226 return False, None, current_bundle_hash
227 else:
228 logger.info(
229 "New bundle found, hash is: {hash}",
230 hash=current_bundle_hash,
231 )
232 prev_bundle_hash = self.bundle_hash
233 self.bundle_hash = current_bundle_hash
234 return (
235 tmp_file_path,
236 prev_bundle_hash,
237 current_bundle_hash,
238 )
239 else:
240 if (
241 self.etag == current_etag
242 ): # validate against bad etag implementation
243 logger.info(
244 "No new bundle, hash is: {hash}",
245 hash=current_etag,
246 )
247 return False, None, current_etag
248 prev_etag = self.etag
249 self.etag = current_etag
250 return tmp_file_path, prev_etag, current_etag
252 except (aiohttp.ClientError, HTTPException) as e:
253 logger.warning("server connection error: {err}", err=repr(e))
254 raise
255 except Exception as e:
256 logger.error("unexpected server connection error: {err}", err=repr(e))
257 raise
259 async def check_for_changes(self):
260 """Calling this method will trigger an api check to the remote.
262 If after the request the watcher detects new bundle, it will
263 call the callbacks registered with _on_new_policy().
264 """
265 logger.info(
266 "Fetching changes from remote: '{remote}'",
267 remote=redact_url(self.remote_source_url),
268 )
269 (
270 has_changes,
271 prev,
272 latest,
273 prev_commit,
274 new_commit,
275 ) = await self.api_update_policy()
276 if not has_changes:
277 logger.info("No new version: current hash is: {head}", head=latest)
278 else:
279 logger.info(
280 "Found new version: old version hash was '{prev_head}', new version hash is '{new_head}'",
281 prev_head=prev,
282 new_head=latest,
283 )
284 await self._on_new_policy(old=prev_commit, new=new_commit)