import functools
import json
from typing import List, Optional, Callable, Union
from q2_sdk.core.exceptions import TectonError
from q2_sdk.core.http_handlers.base_handler import ConfigurationCheck
from q2_sdk.core.http_handlers.tecton_base_handler import (
EdvMfaType,
MFAMessage,
Q2TectonBaseRequestHandler,
_build_mfa_response_payload,
_detect_mfa_exit,
_mfa_handling_supported,
_run_audit_check,
_serialize_for_ajax_if_needed,
)
from q2_sdk.models.tecton import InternalServerError, Success
from q2_sdk.tools import utils, jinja
from q2_sdk.core.cli import colored, textui
from q2_sdk.ui import forms
async def _handle_mfa_exit(handler, error_handler, is_ajax: bool = False):
detected, result = _detect_mfa_exit(handler, error_handler)
if not detected:
return False, None
if result is not None:
return True, _serialize_for_ajax_if_needed(handler, result, is_ajax)
handler.logger.warning(MFAMessage.NO_ERROR_HANDLER.value)
if is_ajax and _mfa_handling_supported(handler):
return True, json.dumps(Success({}).to_json())
return True, await handler.default()
async def _handle_audit_and_response(
handler,
func,
error_handler: Optional[Callable],
audit_action_name: str,
audit_details: Optional[dict | str | Callable],
soft_failures: bool = False,
is_ajax: bool = False,
):
"""
Handle audit record creation and processing the response.
"""
def format_challenge(audit_response):
if is_ajax and _mfa_handling_supported(handler):
# Pre-serialize: @create_audit(...)/mfa_validation_required() is the
# outer decorator in the correct stacking order, so this return value
# never passes through @ajax's own wrapper — nothing downstream will
# serialize it otherwise.
return json.dumps(
Success(_build_mfa_response_payload(audit_response)).to_json()
)
return handler._generate_mfa_html(audit_response)
def format_denial(err_msg, audit_response):
if is_ajax and _mfa_handling_supported(handler):
return json.dumps(
InternalServerError(
"Action Denied",
{
"detail": err_msg,
"errorReturnCode": audit_response.get("return_code"),
},
).to_json()
)
return handler._generate_mfa_html(audit_response)
def wrapped_error_handler(h, err_msg):
return _serialize_for_ajax_if_needed(h, error_handler(h, err_msg), is_ajax)
return await _run_audit_check(
handler,
func,
wrapped_error_handler if error_handler is not None else None,
audit_action_name,
audit_details,
format_challenge=format_challenge,
format_denial=format_denial,
soft_failures=soft_failures,
)
def inner_mfa_function(
func,
error_handler: Optional[Callable] = None,
audit_action_name: Optional[str] = None,
audit_details: Optional[Union[dict, str, Callable]] = None,
mfa_type: Optional[EdvMfaType] = None,
single_use: bool = False,
):
"""
Decorator for checking if session has been validated by the MFA workflow
:param error_handler: Optional callable to handle errors
:param audit_action_name: Optional audit action name for EDV workflow
:param audit_details: Optional audit details for EDV workflow
:param mfa_type: Optional EdvMfaType enum value (HQ defaults to PatrolEDV if not specified)
:param single_use: If True, deletes the cached token after the handler runs, forcing re-validation on the next request
"""
is_ajax = getattr(func, "is_ajax", False)
@functools.wraps(func)
async def wrapper(handler):
validated = handler.validate_mfa_token()
if validated:
handler.logger.debug(MFAMessage.VALIDATED.value)
result = await func(handler)
if single_use:
handler.consume_mfa_token()
return result
else:
handler.logger.debug(MFAMessage.NOT_FOUND.value)
exit_detected, exit_response = await _handle_mfa_exit(
handler, error_handler, is_ajax=is_ajax
)
if exit_detected:
return exit_response
if audit_action_name:
return await _handle_audit_and_response(
handler,
func,
error_handler,
audit_action_name,
audit_details,
is_ajax=is_ajax,
)
else:
mfa_type_response = await handler._get_mfa_type(mfa_type=mfa_type)
if is_ajax and _mfa_handling_supported(handler):
return json.dumps(
Success(_build_mfa_response_payload(mfa_type_response)).to_json()
)
html = handler._generate_mfa_html(mfa_type_response)
return html
return wrapper
def mfa_validation_required(
error_handler: Optional[Callable] = None,
audit_action_name: Optional[str] = None,
audit_details: Optional[Union[dict, str, Callable]] = None,
mfa_type: Optional[EdvMfaType] = None,
single_use: bool = False,
):
"""
Decorator for checking if session has been validated by the MFA workflow
:param error_handler: Optional callable to handle errors
:param audit_action_name: Optional audit action name for EDV workflow
:param audit_details: Optional audit details for EDV workflow
:param mfa_type: Optional EdvMfaType enum value (defaults to PatrolEDV if not specified)
:param single_use: If True, deletes the cached token after the handler runs, forcing re-validation on the next request
"""
return functools.partial(
inner_mfa_function,
error_handler=error_handler,
audit_action_name=audit_action_name,
audit_details=audit_details,
mfa_type=mfa_type,
single_use=single_use,
)
def inner_audit_function(
func,
audit_action_name: str,
audit_details: Optional[dict | str | Callable] = None,
error_handler: Optional[Callable] = None,
csr_only: bool = False,
soft_failures: bool = False,
):
"""
Decorator for creating an audit record and handling edv response as needed
"""
is_ajax = getattr(func, "is_ajax", False)
@functools.wraps(func)
async def wrapper(handler):
if csr_only and handler.online_session.is_csr_assist is False:
return await func(handler)
routing_key = handler.form_fields.get("routing_key", "")
if routing_key in handler._unparsed_routes:
handler.logger.debug(
f"Skipping audit '{audit_action_name}' for unparsed route: {routing_key}"
)
return await func(handler)
exit_detected, exit_response = await _handle_mfa_exit(
handler, error_handler, is_ajax=is_ajax
)
if exit_detected:
return exit_response
return await _handle_audit_and_response(
handler,
func,
error_handler,
audit_action_name,
audit_details,
soft_failures,
is_ajax=is_ajax,
)
return wrapper
def create_audit(
audit_action_name: str,
audit_details: Optional[Union[dict, str, Callable]] = None,
error_handler: Optional[Callable] = None,
csr_only: bool = False,
soft_failures: bool = False,
):
return functools.partial(
inner_audit_function,
error_handler=error_handler,
audit_action_name=audit_action_name,
audit_details=audit_details,
csr_only=csr_only,
soft_failures=soft_failures,
)
[docs]
class Q2TectonServerRequestHandler(Q2TectonBaseRequestHandler):
"""
RequestHandler meant to be used for requests incoming for Tecton Server Side Rendered Content.
"""
def __init__(self, application, request, **kwargs):
self.auth_tokens = []
super().__init__(application, request, **kwargs)
async def mfa_validate(self):
try:
self.return_as_html = False
return await super().mfa_validate()
except TectonError as exc:
return json.dumps(
InternalServerError(
exc.args[0], exc.args[1] if len(exc.args) > 1 else None
).to_json()
)
def _generate_mfa_html(self, response):
replay_fields = {
key: value
for key, value in self.form_fields.items()
if not key.startswith("tct-")
}
replay_fields["q2token"] = self.online_session.session_id
prompt_options = {"code": response.get("return_code")}
if response.get("is_external_mfa"):
prompt_options["externalMfaShortName"] = response.get(
"external_mfa_short_name"
)
html = jinja.jinja_generate(
"q2_mfa_handler.html.jinja2",
{
"extension_name": self.extension_name,
"form_fields": replay_fields,
"prompt_options": prompt_options,
"tecton_url": self.TECTON_URL,
"is_error": response.get("is_edv_denied", False)
or response.get("is_unknown_error", False),
},
)
return html
@classmethod
async def _fix_manifest_url_paths(cls):
from q2_sdk.entrypoints import tecton
conf_file_name = cls.CONFIG_FILE_NAME or cls.extension_name
conf_file = tecton.import_config_file_as_dict(conf_file_name)
manifest = conf_file["FEATURE"]
for mod in manifest["modules"]:
tecton.set_url_paths(
config=manifest,
entrypoint_name=mod,
is_server_side=True,
set_unauth=cls.IS_UNAUTHENTICATED,
set_mfa=cls.IS_MFA,
)
tecton.write_config_file(cls.extension_name, conf_file, log=False)
textui.puts(colored.green(f"Fixes applied to {conf_file_name} config file"))
@classmethod
async def validate_configuration(cls) -> List[ConfigurationCheck]:
config_checks = await super().validate_configuration()
config_checks.append(cls._validate_feature_manifest())
return config_checks
@classmethod
def _validate_feature_manifest(cls) -> ConfigurationCheck:
manifest = utils.get_feature_manifest(cls.extension_name)
configuration_check = ConfigurationCheck(True)
for module in manifest["modules"].values():
url_is_unauth = module["url"].startswith("./sdk/{featureName}")
url_is_mfa = module["url"].startswith("./sdk/mfa/{featureName}")
class_is_unauth = cls.IS_UNAUTHENTICATED
class_is_mfa = cls.IS_MFA
unauth_url_discrepancy = (
class_is_unauth and class_is_unauth != url_is_unauth
)
mfa_url_discrepancy = class_is_mfa and class_is_mfa != url_is_mfa
if unauth_url_discrepancy:
return ConfigurationCheck(
False,
(
"Authentication choice in Class must match Tecton config\n"
f"Class is unauth: {class_is_unauth}\n"
f"Tecton config is unauth: {url_is_unauth}"
),
cls._fix_manifest_url_paths(),
)
if mfa_url_discrepancy:
return ConfigurationCheck(
False,
(
"MFA choice in Class must match Tecton config\n"
f"Class is MFA: {class_is_mfa}\n"
f"Tecton config is MFA: {url_is_mfa}"
),
cls._fix_manifest_url_paths(),
)
return configuration_check
[docs]
@functools.cached_property
def incoming_context(self):
"""
Lazily retrieves incoming context from form fields.
"""
if not hasattr(self, "form_fields") or not self.form_fields:
return {}
context_obj = {}
if "tct-ctxid" in self.form_fields:
context_obj["value"] = self.form_fields["tct-ctxid"]
if "tct-ctxname" in self.form_fields:
context_obj["id"] = self.form_fields["tct-ctxname"]
return context_obj