opal_common/sources/api_policy_source.py
async def fetch_policy_bundle_from_api_source(
        self,
        url: str,
        token: Optional[str]
    ) -> Tuple[Path, BundleHash, BundleHash]:
        """
        Fetches the bundle. May throw, in which case we retry again.
        Checks that the bundle file isn't the same with Etag, if server
        doesn't have Etag it checks it with hash on the bundle file

        Read more on Etag here:
        https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/ETag

        Args:
            url(str): the base address to request the bundle.tar.gz file from
            token (str, optional): Auth token to include in connections to OPAL server. Defaults to POLICY_BUNDLE_SERVER_TOKEN.
        Returns:
            Path: path to the bundle file that we just downloaded from the remote API source
            BundleHash: previous bundle hash on None if this is the initial bundle file
            BundleHash: current bundle hash

        """
        auth_headers = tuple_to_dict(
            get_authorization_header(token)) if token else {}
        etag_headers = {'ETag': self.etag,
                        "If-None-Match": self.etag} if self.etag else {}
        full_url = f"{url}/bundle.tar.gz"

        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(
                    f"{full_url}",
                    headers={'content-type': 'application/gzip',
                             **auth_headers, **etag_headers},
                ) as response:
                    if response.status == status.HTTP_404_NOT_FOUND:
                        logger.warning(
                            "requested url not found: {full_url}", full_url=full_url)
                        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"requested url not found: {full_url}")
                    if response.status == status.HTTP_304_NOT_MODIFIED:
                        logger.info(
                            "Not modified at: {now}", now=datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
                        return False, None, self.etag

                    # may throw ValueError
                    await throw_if_bad_status_code(response, expected=[status.HTTP_200_OK])
                    current_etag = response.headers.get("ETag", None)
                    response_bytes = await response.read()
                    tmp_file_path = Path(opal_server_config.POLICY_BUNDLE_TMP_PATH)
                    with open(tmp_file_path, "wb") as file:
                        file.write(response_bytes)

                    if not current_etag:
                        logger.info('Etag is turnned off, you may want to turn it on at your bundle server')
                        current_bundle_hash = hash_file(tmp_file_path)
                        logger.info('Bundle hash is {hash}', hash=current_bundle_hash)
                        if self.bundle_hash == current_bundle_hash:
                            logger.info("No new bundle, hash is: {hash}", hash=current_bundle_hash)
                            return False, None, current_bundle_hash
                        else:
                            logger.info("New bundle found, hash is: {hash}", hash=current_bundle_hash)
                            prev_bundle_hash = self.bundle_hash
                            self.bundle_hash = current_bundle_hash
                            return tmp_file_path, prev_bundle_hash, current_bundle_hash
                    else:
                        if self.etag == current_etag: # validate against bad etag implementation
                            logger.info("No new bundle, hash is: {hash}", hash=current_etag)
                            return False, None, current_etag
                        prev_etag = self.etag
                        self.etag = current_etag
                        return tmp_file_path, prev_etag, current_etag

            except (aiohttp.ClientError, HTTPException) as e:
                logger.warning("server connection error: {err}", err=repr(e))
                raise
            except Exception as e:
                logger.error("unexpected server connection error: {err}", err=repr(e))
                raise

        Fetches the bundle. May throw, in which case we retry again.
        Checks that the bundle file isn't the same with Etag, if server
        doesn't have Etag it checks it with hash on the bundle file

        Read more on Etag here:
        https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/ETag

        Args:
            url(str): the base address to request the bundle.tar.gz file from
            token (str, optional): Auth token to include in connections to OPAL server. Defaults to POLICY_BUNDLE_SERVER_TOKEN.
        Returns:
            Path: path to the bundle file that we just downloaded from the remote API source
            BundleHash: previous bundle hash on None if this is the initial bundle file
            BundleHash: current bundle hash

        
Fetches the bundle. May throw, in which case we retry again.
Checks that the bundle file isn't the same with Etag, if server
doesn't have Etag it checks it with hash on the bundle file


Fetches the bundle . May throw in which case we retry again . Checks that the bundle file isn ' t the same with Etag if server doesn ' t have Etag it checks it with hash on the bundle file
async def fetch_policy_bundle_from_api_source(
        self,
        url: str,
        token: Optional[str]
    ) -> Tuple[Path, BundleHash, BundleHash]:
        auth_headers = tuple_to_dict(
            get_authorization_header(token)) if token else {}
        etag_headers = {'ETag': self.etag,
                        "If-None-Match": self.etag} if self.etag else {}
        full_url = f"{url}/bundle.tar.gz"
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(
                    f"{full_url}",
                    headers={'content-type': 'application/gzip',
                             **auth_headers, **etag_headers},
                ) as response:
                    if response.status == status.HTTP_404_NOT_FOUND:
                        logger.warning(
                            "requested url not found: {full_url}", full_url=full_url)
                        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"requested url not found: {full_url}")
                    if response.status == status.HTTP_304_NOT_MODIFIED:
                        logger.info(
                            "Not modified at: {now}", now=datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
                        return False, None, self.etag
                    await throw_if_bad_status_code(response, expected=[status.HTTP_200_OK])
                    current_etag = response.headers.get("ETag", None)
                    response_bytes = await response.read()
                    tmp_file_path = Path(opal_server_config.POLICY_BUNDLE_TMP_PATH)
                    with open(tmp_file_path, "wb") as file:
                        file.write(response_bytes)
                    if not current_etag:
                        logger.info('Etag is turnned off, you may want to turn it on at your bundle server')
                        current_bundle_hash = hash_file(tmp_file_path)
                        logger.info('Bundle hash is {hash}', hash=current_bundle_hash)
                        if self.bundle_hash == current_bundle_hash:
                            logger.info("No new bundle, hash is: {hash}", hash=current_bundle_hash)
                            return False, None, current_bundle_hash
                        else:
                            logger.info("New bundle found, hash is: {hash}", hash=current_bundle_hash)
                            prev_bundle_hash = self.bundle_hash
                            self.bundle_hash = current_bundle_hash
                            return tmp_file_path, prev_bundle_hash, current_bundle_hash
                    else:
                        if self.etag == current_etag: 
                            logger.info("No new bundle, hash is: {hash}", hash=current_etag)
                            return False, None, current_etag
                        prev_etag = self.etag
                        self.etag = current_etag
                        return tmp_file_path, prev_etag, current_etag
            except (aiohttp.ClientError, HTTPException) as e:
                logger.warning("server connection error: {err}", err=repr(e))
                raise
            except Exception as e:
                logger.error("unexpected server connection error: {err}", err=repr(e))
                raise
async def fetch_policy_bundle_from_api_source ( self , url : str , token : Optional [ str ] ) -> Tuple [ Path , BundleHash , BundleHash ] : auth_headers = tuple_to_dict ( get_authorization_header ( token ) ) if token else { } etag_headers = { 'ETag' : self . etag , "If-None-Match" : self . etag } if self . etag else { } full_url = f"{url}/bundle.tar.gz" async with aiohttp . ClientSession ( ) as session : try : async with session . get ( f"{full_url}" , headers = { 'content-type' : 'application/gzip' , ** auth_headers , ** etag_headers } , ) as response : if response . status == status . HTTP_404_NOT_FOUND : logger . warning ( "requested url not found: {full_url}" , full_url = full_url ) raise HTTPException ( status_code = status . HTTP_404_NOT_FOUND , detail = f"requested url not found: {full_url}" ) if response . status == status . HTTP_304_NOT_MODIFIED : logger . info ( "Not modified at: {now}" , now = datetime . now ( ) . strftime ( '%Y-%m-%d %H:%M:%S' ) ) return False , None , self . etag await throw_if_bad_status_code ( response , expected = [ status . HTTP_200_OK ] ) current_etag = response . headers . get ( "ETag" , None ) response_bytes = await response . read ( ) tmp_file_path = Path ( opal_server_config . POLICY_BUNDLE_TMP_PATH ) with open ( tmp_file_path , "wb" ) as file : file . write ( response_bytes ) if not current_etag : logger . info ( 'Etag is turnned off, you may want to turn it on at your bundle server' ) current_bundle_hash = hash_file ( tmp_file_path ) logger . info ( 'Bundle hash is {hash}' , hash = current_bundle_hash ) if self . bundle_hash == current_bundle_hash : logger . info ( "No new bundle, hash is: {hash}" , hash = current_bundle_hash ) return False , None , current_bundle_hash else : logger . info ( "New bundle found, hash is: {hash}" , hash = current_bundle_hash ) prev_bundle_hash = self . bundle_hash self . bundle_hash = current_bundle_hash return tmp_file_path , prev_bundle_hash , current_bundle_hash else : if self . etag == current_etag : logger . info ( "No new bundle, hash is: {hash}" , hash = current_etag ) return False , None , current_etag prev_etag = self . etag self . etag = current_etag return tmp_file_path , prev_etag , current_etag except ( aiohttp . ClientError , HTTPException ) as e : logger . warning ( "server connection error: {err}" , err = repr ( e ) ) raise except Exception as e : logger . error ( "unexpected server connection error: {err}" , err = repr ( e ) ) raise
Fetches the bundle.
Fetches the bundle .
"""
        Fetches the bundle. May throw, in which case we retry again.
        Checks that the bundle file isn't the same with Etag, if server
        doesn't have Etag it checks it with hash on the bundle file

        Read more on Etag here:
        https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/ETag

        Args:
            url(str): the base address to request the bundle.tar.gz file from
            token (str, optional): Auth token to include in connections to OPAL server. Defaults to POLICY_BUNDLE_SERVER_TOKEN.
        Returns:
            Path: path to the bundle file that we just downloaded from the remote API source
            BundleHash: previous bundle hash on None if this is the initial bundle file
            BundleHash: current bundle hash

        """ # may throw ValueError # validate against bad etag implementation
path to the bundle file that we just downloaded from the remote API source
BundleHash: previous bundle hash on None if this is the initial bundle file
BundleHash: current bundle hash
path to the bundle file that we just downloaded from the remote API source BundleHash : previous bundle hash on None if this is the initial bundle file BundleHash : current bundle hash
Path
self
url
str
the base address to request the bundle.tar.gz file from
the base address to request the bundle . tar . gz file from
token
Optional[str]
Auth token to include in connections to OPAL server. Defaults to POLICY_BUNDLE_SERVER_TOKEN.
Auth token to include in connections to OPAL server . Defaults to POLICY_BUNDLE_SERVER_TOKEN .
POLICY_BUNDLE_SERVER_TOKEN