Source code for q2_sdk.core.install_steps.uux_payload

import base64
from typing import Unpack, Optional
import uuid

from dataclasses import dataclass, field
from io import StringIO
from json import loads, dumps
from lxml import etree
from lxml.etree import _ElementUnicodeResult

from q2_sdk.core.exceptions import ConfigurationError
from q2_sdk.core.install_steps.base import InstallStep, InstallStepArguments
from q2_sdk.hq.models.online_session import OnlineSession
from q2_sdk.hq.models.online_user import OnlineUser
from q2_sdk.hq.db.ui_config_property_data import UiConfigPropertyData
from q2_sdk.hq.db.wedge_address import WedgeAddress
from q2_sdk.tools import utils, jinja


[docs] @dataclass class UUXPayloadScript: """Dataclass used to define the scripts to be included in the uux.aspx payload. These should be passed in as a List to the UUXPayload constructor. """ #: url if sdk_asset = False or filename (in frontend/public folder) if sdk_asset is True of the javascript file to include url: str #: https://developer.mozilla.org/en-US/docs/Web/HTML/Element/script#attr-defer defer_load: bool = False #: https://developer.mozilla.org/en-US/docs/Web/HTML/Element/script#attr-integrity. Required if sdk_asset false. integrity: Optional[str] = None #: Indicates if the file referenced by `url` is local to this extension sdk_asset: bool = True #: List of WEDGE_ADDRESS_CONFIGS keys to include as variables in JS. Will be placed in window.__{extension_name} dictionary wedge_address_js_variables: list = field(default_factory=lambda: []) def __json__(self): return vars(self)
[docs] class UUXPayload(InstallStep): """Install a JS file for inclusion in uux.aspx payload""" PROPERTY_NAME = "Q2_SECURITY_SECURITY_THIRD_PARTY_HTML_BLOCK" def __init__( self, scripts: list[UUXPayloadScript], base_override=None, **kwargs: Unpack[InstallStepArguments], ): """ The UUXPayload class takes a list of UUXPayloadScript objects. They will be placed sequentially into the bottom of the uux.aspx's HTML. :param scripts: List of UUXPayloadScript objects :param base_override: value to use as the base asset url instead of using the url of the running container """ kwargs["allow_editable"] = False super().__init__(**kwargs) if isinstance(scripts, UUXPayloadScript): scripts = [scripts] if all([isinstance(x, dict) for x in scripts]): scripts = [ UUXPayloadScript( x["url"], defer_load=x["defer_load"], integrity=x["integrity"], sdk_asset=x["sdk_asset"], wedge_address_js_variables=x.get("wedge_address_js_variables", []), ) for x in scripts ] for script in scripts: if not isinstance(script, UUXPayloadScript): raise ConfigurationError( "scripts parameter must be filled with UUXPayloadScript objects" ) self.scripts = scripts self.base_override = base_override self.not_install_var_list.append("base_override") def __json__(self): return {**self.install_vars, **{"scripts": self.scripts}} def __repr__(self): return str({"scripts": self.scripts})
[docs] async def install(self): await super().install() wa_js_string = "" wa_js_dict = {} new_payload = "" existing_payload = "" wedge_address_config = {} ui_config = UiConfigPropertyData( self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True ) uiconfig_result = await ui_config.get(self.PROPERTY_NAME) if uiconfig_result: property_value = uiconfig_result[0].findtext("PropertyValue") existing_payload = property_value or "<head></head>" base = f"{utils.get_base_assets_url(self.extension_name, False)}" if self.base_override: base = self.base_override for script in self.scripts: if script.sdk_asset: script.url = f"{base}/{script.url}" elif script.integrity is None: raise ConfigurationError( "Any script sourced from a third party site must include " "integrity hash. " "https://developer.mozilla.org/en-US/docs/Web/Security/" "Subresource_Integrity" ) if len(script.wedge_address_js_variables) > 0: if not wedge_address_config: wedge_address = WedgeAddress( self.logger, hq_credentials=self.hq_credentials ) wa_result = await wedge_address.get(self.extension_name) if wa_result: wedge_address_config = loads(wa_result[0].Config.text) if not wedge_address_config: raise ConfigurationError( "wedge_address_js_variables set but no wedge address configs in DB" ) else: raise ConfigurationError( "wedge_address_js_variables set but no wedge address configs in DB" ) for wa_variable in script.wedge_address_js_variables: if wa_variable in wedge_address_config: wa_js_dict[wa_variable] = wedge_address_config[wa_variable] else: raise ConfigurationError( f"{wa_variable} not in wedge address configs in DB" ) wa_js_string = dumps(wa_js_dict, separators=(",", ":")) replacements = { "form_name": self.extension_name, "wedge_address_js_string": wa_js_string, "sdk_asset": script.sdk_asset, "url": script.url, "integrity": script.integrity, "defer_load": script.defer_load, } template = "js/uux_payload.js.jinja2" script_payload = jinja.jinja_generate( template, replacements_dict=replacements ) script_payload = "".join(script_payload.split("\n")).encode() new_payload += ( f'<script src="data:text/plain;base64,{base64.b64encode(script_payload).decode()}" ' f'data-q2sdk="{self.extension_name}"></script>' ) if uiconfig_result: session = OnlineSession() session.session_id = str(uuid.uuid4()) session.workstation = str(uuid.uuid4()) online_user = OnlineUser() online_user.customer_id = 0 online_user.user_id = 0 online_user.user_logon_id = 0 parser = etree.HTMLParser() tree = etree.parse(StringIO(existing_payload), parser) # remove any existing script tags that have the same fingerprint for old in tree.xpath(f'//script[@data-q2sdk="{self.extension_name}"]'): old.getparent().remove(old) # add new script payload to beginning of cleansed payload for node in tree.xpath("//head/node()"): try: if isinstance(node, str): new_payload += node else: new_payload += etree.tostring( node, encoding="unicode", method="html" ) except TypeError: self.logger.info( "failed to parse existing %s value", self.PROPERTY_NAME ) break return await ui_config.update( self.PROPERTY_NAME, new_payload, session, online_user, ui_source="OnlineBanking", ) else: return await ui_config.create(self.PROPERTY_NAME, "string", new_payload)
[docs] async def uninstall(self): existing_payload = None ui_config = UiConfigPropertyData( self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True ) uiconfig_result = await ui_config.get(self.PROPERTY_NAME) if uiconfig_result: existing_payload = ( f"<head>{uiconfig_result[0].findtext('PropertyValue', '')}</head>" ) if uiconfig_result: cleansed_payload = "" session = OnlineSession() session.session_id = str(uuid.uuid4()) session.workstation = str(uuid.uuid4()) online_user = OnlineUser() online_user.customer_id = 0 online_user.user_id = 0 online_user.user_logon_id = 0 parser = etree.HTMLParser() tree = etree.parse(StringIO(existing_payload), parser) # remove any existing script tags that have the same fingerprint for old in tree.xpath(f'//script[@data-q2sdk="{self.extension_name}"]'): old.getparent().remove(old) # add new script payload to end of cleansed payload cleansed_payload = "" if tree.xpath("//head/node()"): for cleansed_node in tree.xpath("//head/node()"): try: cleansed_payload += etree.tostring( cleansed_node, encoding="unicode", method="html" ) except TypeError: if isinstance(cleansed_node, _ElementUnicodeResult): pass # these seem to be newlines that are not able to be parsed as valid XML or serialized as a string else: raise return await ui_config.update( self.PROPERTY_NAME, cleansed_payload, session, online_user, ui_source="OnlineBanking", ) return None