30004242229

Committed 23 Jul 2026 11:44AM UTC coverage: 91.402% (-0.01%) from 91.413%

Build # 30004242229

Build Type

Pull #2410

github

Committed by

web-flow

Commit Message

Merge 2f6105980 into 2576f4e5d

Pull Request Pull Request #2410: Async callback response

Coverage Stats

8 of 8 new or added lines in 1 file covered. (100.0%)

4 existing lines in 2 files now uncovered.

31435 of 34392 relevant lines covered (91.4%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.47

/kairon/shared/callback/data_objects.py

import base64
import time
from datetime import datetime
from enum import Enum
from typing import Any, Optional, Dict
import json

from pydantic import BaseModel
from uuid6 import uuid7

from mongoengine import StringField, DictField, DateTimeField, Document, DynamicField, IntField, BooleanField, \
    FloatField

from kairon import Utility
from kairon.async_callback.exceptions import CallbackException
from kairon.exceptions import AppException
from kairon.shared.actions.data_objects import CallbackActionConfig
from kairon.shared.constants import EventClass
from kairon.shared.data.audit.data_objects import Auditlog
from kairon.shared.data.signals import push_notification
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend


def check_nonempty_string(value, msg="Value must be a non-empty string"):
    if not isinstance(value, str) or not value:
        raise AppException(msg)


def encrypt_secret(secret: str) -> str:
    secret = secret.encode("utf-8")
    fernet = Fernet(Utility.environment['security']['fernet_key'].encode("utf-8"))
    return fernet.encrypt(secret).decode("utf-8")


def decrypt_secret(encrypted_secret: str) -> str:
    fernet = Fernet(Utility.environment['security']['fernet_key'].encode("utf-8"))
    return fernet.decrypt(encrypted_secret.encode("utf-8")).decode("utf-8")


def xor_encrypt_secret(secret: str) -> str:
    """
    AES small length text encryption
    TODO: change function name
    """
    key = Utility.environment['async_callback_action']['short_secret']['aes_key']
    iv = Utility.environment['async_callback_action']['short_secret']['aes_iv']
    key = bytes.fromhex(key)
    iv = bytes.fromhex(iv)
    secret_bytes = secret.encode()
    cipher = Cipher(algorithms.AES(key), modes.CTR(iv), backend=default_backend())
    encryptor = cipher.encryptor()
    ciphertext = encryptor.update(secret_bytes) + encryptor.finalize()
    encoded_result = base64.urlsafe_b64encode(ciphertext).decode().rstrip("=")
    return encoded_result


def xor_decrypt_secret(encoded_secret: str) -> str:
    """
    AES encripted text decription function
    TODO: change function name
    """
    key = Utility.environment['async_callback_action']['short_secret']['aes_key']
    iv = Utility.environment['async_callback_action']['short_secret']['aes_iv']
    key = bytes.fromhex(key)
    iv = bytes.fromhex(iv)
    secret = None
    try:
        decoded_secret = base64.urlsafe_b64decode(encoded_secret + "=" * (4 - len(encoded_secret) % 4))
        cipher = Cipher(algorithms.AES(key), modes.CTR(iv), backend=default_backend())
        decryptor = cipher.decryptor()
        secret = decryptor.update(decoded_secret) + decryptor.finalize()
        return secret.decode()
    except Exception:
        raise AppException("Invalid token!")


class CallbackExecutionMode(str, Enum):
    ASYNC = "async"
    SYNC = "sync"

class CallbackResponseType(str, Enum):
    KAIRON_JSON = "kairon_json"
    JSON = "json"
    TEXT = "text"


@push_notification.apply
class CallbackConfig(Auditlog):
    name = StringField(required=True)
    pyscript_code = StringField(required=True)
    validation_secret = StringField(required=True)
    execution_mode = StringField(default=CallbackExecutionMode.ASYNC.value,
                                 choices=[v.value for v in CallbackExecutionMode])
    expire_in = IntField(default=0)
    shorten_token = BooleanField(default=False)
    token_hash = StringField()
    token_value = StringField()
    standalone = BooleanField(default=False)
    standalone_id_path = StringField(default='')
    response_type = StringField(default=CallbackResponseType.KAIRON_JSON.value,
                                choices=[v.value for v in CallbackResponseType])
    bot = StringField(required=True)
    meta = {"indexes": [{"fields": ["bot", "name"]}]}

    @staticmethod
    def get_all_names(bot) -> list[str]:
        names = CallbackConfig.objects(bot=bot).distinct(field="name")
        return list(names)

    @staticmethod
    def get_entry(bot :str, name :str) -> dict:
        entry = CallbackConfig.objects(bot=bot, name__iexact=name).first()
        if not entry:
            raise AppException(f"Callback Configuration with name '{name}' does not exist!")
        dict_form = entry.to_mongo().to_dict()
        dict_form.pop("_id")
        return dict_form

    @staticmethod
    def create_entry(bot: str,
                     name: str,
                     pyscript_code: str,
                     execution_mode: str = CallbackExecutionMode.ASYNC.value,
                     expire_in: int = 30,
                     shorten_token: bool = False,
                     standalone: bool = False,
                     standalone_id_path: str = '',
                     response_type: str = CallbackResponseType.KAIRON_JSON.value,
                     **kwargs):
        check_nonempty_string(name)
        if standalone and not standalone_id_path:
            raise AppException("Standalone ID path is required for standalone callbacks!")
        Utility.is_exist(
            CallbackConfig,
            exp_message=f"Callback Configuration with name '{name}' exists!",
            name__iexact=name,
            bot=bot,
            raise_error=True
        )
        check_nonempty_string(pyscript_code)
        validation_secret = encrypt_secret(uuid7().hex)
        token_hash = None
        if shorten_token:
            token_hash = uuid7().hex
        config = CallbackConfig(name=name,
                                bot=bot,
                                pyscript_code=pyscript_code,
                                validation_secret=validation_secret,
                                execution_mode=execution_mode,
                                expire_in=expire_in,
                                shorten_token=shorten_token,
                                token_hash=token_hash,
                                standalone=standalone,
                                standalone_id_path=standalone_id_path,
                                response_type=response_type,
                                **kwargs)
        config.save()
        return config.to_mongo().to_dict()

    @staticmethod
    def get_auth_token(bot: str, name: str) -> tuple[str, bool]:
        entry = CallbackConfig.objects(bot=bot, name__iexact=name).first()
        if not entry:
            raise AppException(f"Callback Configuration with name '{name}' does not exist!")

        info = {
            "bot": entry.bot,
            "callback_name": entry.name,
            "validation_secret": decrypt_secret(entry.validation_secret),
            "expire_in": entry.expire_in,
        }

        token = encrypt_secret(json.dumps(info))

        if entry.shorten_token:
            entry.token_value = token
            entry.save()
            return xor_encrypt_secret(entry.token_hash), entry.standalone
        else:
            return token, entry.standalone

    @staticmethod
    def verify_auth_token(token: str):
        info = None
        if len(token) < 64:
            search_key = xor_decrypt_secret(token)
            config = CallbackConfig.objects(token_hash=search_key, shorten_token=True).first()
            if config:
                info = json.loads(decrypt_secret(config.token_value))
        else:
            info = json.loads(decrypt_secret(token))
        if not info:
            raise AppException("Invalid token!")

        config = CallbackConfig.objects(bot=info['bot'],
                                        name__iexact=info['callback_name'],
                                        ).first()
        if not config:
            raise AppException("Invalid token!")
        if decrypt_secret(config.validation_secret) != info['validation_secret']:
            raise AppException("Invalid token!")
        return config

    @staticmethod
    def edit(bot: str, name: str, **kwargs):
        check_nonempty_string(name)
        config = CallbackConfig.objects(bot=bot, name__iexact=name).first()
        if not config:
            raise AppException(f"Callback Configuration with name '{name}' does not exist!")
        for key, value in kwargs.items():
            setattr(config, key, value)
        if config.shorten_token and not config.token_hash:
            config.token_hash = uuid7().hex
        config.save()
        return config.to_mongo().to_dict()

    @staticmethod
    def delete_entry(bot: str, name: str):
        check_nonempty_string(name)
        callback_action = CallbackActionConfig.objects(bot=bot, callback_name=name).first()
        if callback_action:
            raise AppException(f"Cannot delete Callback Configuration '{name}' as it is attached to {callback_action.name} callback action!")
        config = CallbackConfig.objects(bot=bot, name__iexact=name).first()
        if not config:
            raise AppException(f"Callback Configuration with name '{name}' does not exist!")
        config.delete()
        return config.name

    @staticmethod
    def get_callback_url(bot: str, name: str):
        base_url = Utility.environment['async_callback_action']['url']
        auth_token, is_standalone = CallbackConfig.get_auth_token(bot, name)
        if not is_standalone:
            raise AppException(f"Callback Configuration with name '{name}' is not standalone!")
        callback_url = f"{base_url}/s/{auth_token}"
        return callback_url


class CallbackRecordStatusType(Enum):
    SUCCESS = "Success"
    FAILED = "Failed"


@push_notification.apply
class CallbackData(Document):
    """
    this represents a record of every callback execution generated by action trigger
    """
    action_name = StringField()
    callback_name = StringField(required=True)
    bot = StringField(required=True)
    sender_id = StringField(required=True)
    channel = StringField(required=True)
    metadata = DictField()
    identifier = StringField(required=True)
    timestamp = FloatField(default=time.time)
    callback_url = StringField()
    execution_mode = StringField(default=CallbackExecutionMode.ASYNC.value,
                                 choices=[v.value for v in CallbackExecutionMode.__members__.values()])
    state = DynamicField(default=0)
    is_valid = BooleanField(default=True)
    meta = {"indexes": [{"fields": ["bot", "identifier"]}]}

    @staticmethod
    def create_entry(name: str, callback_config_name: str, bot: str, sender_id: str, channel: str, metadata: dict, **kwargs):
        check_nonempty_string(name)
        check_nonempty_string(callback_config_name)
        check_nonempty_string(bot)
        check_nonempty_string(sender_id)
        check_nonempty_string(channel)
        identifier = f"{uuid7().hex}"
        base_url = Utility.environment['async_callback_action']['url']
        auth_token, is_standalone = CallbackConfig.get_auth_token(bot, callback_config_name)
        callback_url = f"{base_url}/"
        if is_standalone:
            callback_url += f"s/{auth_token}"
        else:
            callback_url += f"d/{identifier}/{auth_token}"

        record = CallbackData(action_name=name,
                              callback_name=callback_config_name,
                              bot=bot,
                              sender_id=sender_id,
                              channel=channel,
                              metadata=metadata,
                              identifier=identifier,
                              callback_url=callback_url,
                              timestamp=time.time(),
                              is_valid=True,
                              **kwargs)
        record.save()
        return callback_url, identifier, is_standalone

    @staticmethod
    def get_value_from_json(json_obj: Any, path: str):
        keys = path.split('.')
        value = json_obj
        try:
            for key in keys:
                if isinstance(value, list):
                    key = int(key)
                value = value[key]
        except (KeyError, IndexError, ValueError, TypeError):
            raise CallbackException(f"Cannot find identifier at path '{path}' in request data!", 200)

        return value

    @staticmethod
    def validate_entry(token: str, identifier: Optional[str] = None, request_body: Any = None) -> tuple[dict, dict]:
        check_nonempty_string(token)
        config_entry = CallbackConfig.verify_auth_token(token)

        if config_entry.standalone:
            if not request_body:
                raise AppException("Request data is required for standalone callbacks!")
            identifier = CallbackData.get_value_from_json(request_body, config_entry.standalone_id_path)

        record = CallbackData.objects(bot=config_entry.bot, identifier=identifier).first()
        if not record:
            raise AppException("Callback Record does not exist, invalid identifier!")
        if not record.is_valid:
            raise AppException("Callback has been invalidated!")
        if config_entry.expire_in > 0:
            exp_time = record.timestamp + config_entry.expire_in
            if exp_time < time.time():
                raise AppException("Callback time-limit expired")
        entry_dict = record.to_mongo().to_dict()
        entry_dict.pop('_id')
        entry_dict.pop('timestamp')
        callback_dict = config_entry.to_mongo().to_dict()
        return entry_dict, callback_dict

    @staticmethod
    def update_state(bot: str, identifier: str, state: dict, invalidate: bool):
        record = CallbackData.objects(bot=bot, identifier=identifier).first()
        if not record:
            raise AppException("Callback Record does not exist, invalid identifier!")
        record.state = state
        record.is_valid = not invalidate
        record.save()
        return record.to_mongo().to_dict()


@push_notification.apply
class CallbackLog(Document):
    """
        this represents the record of actual execution record of  callback after the callback url is triggered
    """
    callback_name = StringField(required=True)
    bot = StringField(required=True)
    channel = StringField(default='unsupported')
    identifier = StringField(required=True)
    pyscript_code = StringField(required=True)
    sender_id = StringField()
    log = StringField()
    timestamp = DateTimeField(default=datetime.utcnow)
    status = StringField(default=CallbackRecordStatusType.SUCCESS.value,
                         choices=[v.value for v in CallbackRecordStatusType.__members__.values()])
    request_data = DynamicField()
    metadata = DynamicField()
    callback_url = StringField(required=True)
    callback_source = StringField()

    meta = {"indexes": [{"fields": ["bot", "identifier"]}]}

    @staticmethod
    def create_success_entry(name: str,
                             bot: str,
                             channel: str,
                             identifier: str,
                             pyscript_code: str,
                             sender_id: str,
                             log: str,
                             request_data: Any,
                             metadata: dict,
                             callback_url: str,
                             callback_source: str) -> dict:
        check_nonempty_string(name)
        check_nonempty_string(bot)
        check_nonempty_string(identifier)
        check_nonempty_string(pyscript_code)
        check_nonempty_string(callback_url)
        record = CallbackLog(callback_name=name,
                             bot=bot,
                             channel=channel,
                             identifier=identifier,
                             pyscript_code=pyscript_code,
                             sender_id=sender_id,
                             log=log,
                             request_data=request_data,
                             metadata=metadata,
                             callback_url=callback_url,
                             callback_source=callback_source,
                             status=CallbackRecordStatusType.SUCCESS.value,
                             timestamp=datetime.utcnow())
        record.save()
        return record.to_mongo().to_dict()

    @staticmethod
    def create_failure_entry(name: str,
                             bot: str,
                             channel: str,
                             identifier: str,
                             pyscript_code: str,
                             sender_id: str,
                             error_log: str,
                             request_data: Any,
                             metadata: dict,
                             callback_url: str,
                             callback_source: str) -> dict:
        check_nonempty_string(name)
        check_nonempty_string(bot)
        check_nonempty_string(identifier)
        check_nonempty_string(pyscript_code)
        check_nonempty_string(callback_url)
        record = CallbackLog(callback_name=name,
                             bot=bot,
                             channel=channel,
                             identifier=identifier,
                             pyscript_code=pyscript_code,
                             sender_id=sender_id,
                             log=error_log,
                             request_data=request_data,
                             metadata=metadata,
                             callback_url=callback_url,
                             callback_source=callback_source,
                             status=CallbackRecordStatusType.FAILED.value,
                             timestamp=datetime.utcnow())
        record.save()
        return record.to_mongo().to_dict()

    @staticmethod
    def get_logs(query: dict, offset: int, limit: int) -> tuple[list[dict], int]:
        logs = CallbackLog.objects(**query).skip(offset).limit(limit).exclude('id').order_by('-timestamp').to_json()
        logs_dict_list = json.loads(logs)
        for log in logs_dict_list:
            log['timestamp'] = log['timestamp']['$date']
        total = CallbackLog.objects(**query).count()
        return logs_dict_list, total

class PyscriptPayload(BaseModel):
    """
    Incoming JSON payload for restricted-Python execution.
    """
    source_code: str
    predefined_objects: Optional[Dict[str, Any]] = None


class CallbackRequest(BaseModel):
    event_class: EventClass
    data: dict
    task_type: str

1	import base64	1✔
2	import time	1✔
3	from datetime import datetime	1✔
4	from enum import Enum	1✔
5	from typing import Any, Optional, Dict	1✔
6	import json	1✔
7
8	from pydantic import BaseModel	1✔
9	from uuid6 import uuid7	1✔
10
11	from mongoengine import StringField, DictField, DateTimeField, Document, DynamicField, IntField, BooleanField, \	1✔
12	FloatField
13
14	from kairon import Utility	1✔
15	from kairon.async_callback.exceptions import CallbackException	1✔
16	from kairon.exceptions import AppException	1✔
17	from kairon.shared.actions.data_objects import CallbackActionConfig	1✔
18	from kairon.shared.constants import EventClass	1✔
19	from kairon.shared.data.audit.data_objects import Auditlog	1✔
20	from kairon.shared.data.signals import push_notification	1✔
21	from cryptography.fernet import Fernet	1✔
22	from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes	1✔
23	from cryptography.hazmat.backends import default_backend	1✔
24
25
26	def check_nonempty_string(value, msg="Value must be a non-empty string"):	1✔
27	if not isinstance(value, str) or not value:	1✔
28	raise AppException(msg)	1✔
29
30
31	def encrypt_secret(secret: str) -> str:	1✔
32	secret = secret.encode("utf-8")	1✔
33	fernet = Fernet(Utility.environment['security']['fernet_key'].encode("utf-8"))	1✔
34	return fernet.encrypt(secret).decode("utf-8")	1✔
35
36
37	def decrypt_secret(encrypted_secret: str) -> str:	1✔
38	fernet = Fernet(Utility.environment['security']['fernet_key'].encode("utf-8"))	1✔
39	return fernet.decrypt(encrypted_secret.encode("utf-8")).decode("utf-8")	1✔
40
41
42	def xor_encrypt_secret(secret: str) -> str:	1✔
43	"""
44	AES small length text encryption
45	TODO: change function name
46	"""
47	key = Utility.environment['async_callback_action']['short_secret']['aes_key']	1✔
48	iv = Utility.environment['async_callback_action']['short_secret']['aes_iv']	1✔
49	key = bytes.fromhex(key)	1✔
50	iv = bytes.fromhex(iv)	1✔
51	secret_bytes = secret.encode()	1✔
52	cipher = Cipher(algorithms.AES(key), modes.CTR(iv), backend=default_backend())	1✔
53	encryptor = cipher.encryptor()	1✔
54	ciphertext = encryptor.update(secret_bytes) + encryptor.finalize()	1✔
55	encoded_result = base64.urlsafe_b64encode(ciphertext).decode().rstrip("=")	1✔
56	return encoded_result	1✔
57
58
59	def xor_decrypt_secret(encoded_secret: str) -> str:	1✔
60	"""
61	AES encripted text decription function
62	TODO: change function name
63	"""
64	key = Utility.environment['async_callback_action']['short_secret']['aes_key']	1✔
65	iv = Utility.environment['async_callback_action']['short_secret']['aes_iv']	1✔
66	key = bytes.fromhex(key)	1✔
67	iv = bytes.fromhex(iv)	1✔
68	secret = None	1✔
69	try:	1✔
70	decoded_secret = base64.urlsafe_b64decode(encoded_secret + "=" * (4 - len(encoded_secret) % 4))	1✔
71	cipher = Cipher(algorithms.AES(key), modes.CTR(iv), backend=default_backend())	1✔
72	decryptor = cipher.decryptor()	1✔
73	secret = decryptor.update(decoded_secret) + decryptor.finalize()	1✔
74	return secret.decode()	1✔
75	except Exception:	1✔
76	raise AppException("Invalid token!")	1✔
77
78
79	class CallbackExecutionMode(str, Enum):	1✔
80	ASYNC = "async"	1✔
81	SYNC = "sync"	1✔
82
83	class CallbackResponseType(str, Enum):	1✔
84	KAIRON_JSON = "kairon_json"	1✔
85	JSON = "json"	1✔
86	TEXT = "text"	1✔
87
88
89	@push_notification.apply	1✔
90	class CallbackConfig(Auditlog):	1✔
91	name = StringField(required=True)	1✔
92	pyscript_code = StringField(required=True)	1✔
93	validation_secret = StringField(required=True)	1✔
94	execution_mode = StringField(default=CallbackExecutionMode.ASYNC.value,	1✔
95	choices=[v.value for v in CallbackExecutionMode])
96	expire_in = IntField(default=0)	1✔
97	shorten_token = BooleanField(default=False)	1✔
98	token_hash = StringField()	1✔
99	token_value = StringField()	1✔
100	standalone = BooleanField(default=False)	1✔
101	standalone_id_path = StringField(default='')	1✔
102	response_type = StringField(default=CallbackResponseType.KAIRON_JSON.value,	1✔
103	choices=[v.value for v in CallbackResponseType])
104	bot = StringField(required=True)	1✔
105	meta = {"indexes": [{"fields": ["bot", "name"]}]}	1✔
106
107	@staticmethod	1✔
108	def get_all_names(bot) -> list[str]:	1✔
109	names = CallbackConfig.objects(bot=bot).distinct(field="name")	1✔
110	return list(names)	1✔
111
112	@staticmethod	1✔
113	def get_entry(bot :str, name :str) -> dict:	1✔
114	entry = CallbackConfig.objects(bot=bot, name__iexact=name).first()	1✔
115	if not entry:	1✔
116	raise AppException(f"Callback Configuration with name '{name}' does not exist!")	1✔
117	dict_form = entry.to_mongo().to_dict()	1✔
118	dict_form.pop("_id")	1✔
119	return dict_form	1✔
120
121	@staticmethod	1✔
122	def create_entry(bot: str,	1✔
123	name: str,
124	pyscript_code: str,
125	execution_mode: str = CallbackExecutionMode.ASYNC.value,
126	expire_in: int = 30,
127	shorten_token: bool = False,
128	standalone: bool = False,
129	standalone_id_path: str = '',
130	response_type: str = CallbackResponseType.KAIRON_JSON.value,
131	**kwargs):
132	check_nonempty_string(name)	1✔
133	if standalone and not standalone_id_path:	1✔
134	raise AppException("Standalone ID path is required for standalone callbacks!")	×
135	Utility.is_exist(	1✔
136	CallbackConfig,
137	exp_message=f"Callback Configuration with name '{name}' exists!",
138	name__iexact=name,
139	bot=bot,
140	raise_error=True
141	)
142	check_nonempty_string(pyscript_code)	1✔
143	validation_secret = encrypt_secret(uuid7().hex)	1✔
144	token_hash = None	1✔
145	if shorten_token:	1✔
146	token_hash = uuid7().hex	×
147	config = CallbackConfig(name=name,	1✔
148	bot=bot,
149	pyscript_code=pyscript_code,
150	validation_secret=validation_secret,
151	execution_mode=execution_mode,
152	expire_in=expire_in,
153	shorten_token=shorten_token,
154	token_hash=token_hash,
155	standalone=standalone,
156	standalone_id_path=standalone_id_path,
157	response_type=response_type,
158	**kwargs)
159	config.save()	1✔
160	return config.to_mongo().to_dict()	1✔
161
162	@staticmethod	1✔
163	def get_auth_token(bot: str, name: str) -> tuple[str, bool]:	1✔
164	entry = CallbackConfig.objects(bot=bot, name__iexact=name).first()	1✔
165	if not entry:	1✔
166	raise AppException(f"Callback Configuration with name '{name}' does not exist!")	1✔
167
168	info = {	1✔
169	"bot": entry.bot,
170	"callback_name": entry.name,
171	"validation_secret": decrypt_secret(entry.validation_secret),
172	"expire_in": entry.expire_in,
173	}
174
175	token = encrypt_secret(json.dumps(info))	1✔
176
177	if entry.shorten_token:	1✔
178	entry.token_value = token	×
179	entry.save()	×
180	return xor_encrypt_secret(entry.token_hash), entry.standalone	×
181	else:
182	return token, entry.standalone	1✔
183
184	@staticmethod	1✔
185	def verify_auth_token(token: str):	1✔
186	info = None	1✔
187	if len(token) < 64:	1✔
188	search_key = xor_decrypt_secret(token)	1✔
189	config = CallbackConfig.objects(token_hash=search_key, shorten_token=True).first()	1✔
190	if config:	1✔
191	info = json.loads(decrypt_secret(config.token_value))	1✔
192	else:
193	info = json.loads(decrypt_secret(token))	1✔
194	if not info:	1✔
195	raise AppException("Invalid token!")	×
196
197	config = CallbackConfig.objects(bot=info['bot'],	1✔
198	name__iexact=info['callback_name'],
199	).first()
200	if not config:	1✔
201	raise AppException("Invalid token!")	×
202	if decrypt_secret(config.validation_secret) != info['validation_secret']:	1✔
203	raise AppException("Invalid token!")	×
204	return config	1✔
205
206	@staticmethod	1✔
207	def edit(bot: str, name: str, **kwargs):	1✔
208	check_nonempty_string(name)	1✔
209	config = CallbackConfig.objects(bot=bot, name__iexact=name).first()	1✔
210	if not config:	1✔
211	raise AppException(f"Callback Configuration with name '{name}' does not exist!")	×
212	for key, value in kwargs.items():	1✔
213	setattr(config, key, value)	1✔
214	if config.shorten_token and not config.token_hash:	1✔
215	config.token_hash = uuid7().hex	×
216	config.save()	1✔
217	return config.to_mongo().to_dict()	1✔
218
219	@staticmethod	1✔
220	def delete_entry(bot: str, name: str):	1✔
221	check_nonempty_string(name)	1✔
222	callback_action = CallbackActionConfig.objects(bot=bot, callback_name=name).first()	1✔
223	if callback_action:	1✔
224	raise AppException(f"Cannot delete Callback Configuration '{name}' as it is attached to {callback_action.name} callback action!")	×
225	config = CallbackConfig.objects(bot=bot, name__iexact=name).first()	1✔
226	if not config:	1✔
227	raise AppException(f"Callback Configuration with name '{name}' does not exist!")	×
228	config.delete()	1✔
229	return config.name	1✔
230
231	@staticmethod	1✔
232	def get_callback_url(bot: str, name: str):	1✔
233	base_url = Utility.environment['async_callback_action']['url']	1✔
234	auth_token, is_standalone = CallbackConfig.get_auth_token(bot, name)	1✔
235	if not is_standalone:	1✔
236	raise AppException(f"Callback Configuration with name '{name}' is not standalone!")	×
237	callback_url = f"{base_url}/s/{auth_token}"	1✔
238	return callback_url	1✔
239
240
241	class CallbackRecordStatusType(Enum):	1✔
242	SUCCESS = "Success"	1✔
243	FAILED = "Failed"	1✔
244
245
246	@push_notification.apply	1✔
247	class CallbackData(Document):	1✔
248	"""
249	this represents a record of every callback execution generated by action trigger
250	"""
251	action_name = StringField()	1✔
252	callback_name = StringField(required=True)	1✔
253	bot = StringField(required=True)	1✔
254	sender_id = StringField(required=True)	1✔
255	channel = StringField(required=True)	1✔
256	metadata = DictField()	1✔
257	identifier = StringField(required=True)	1✔
258	timestamp = FloatField(default=time.time)	1✔
259	callback_url = StringField()	1✔
260	execution_mode = StringField(default=CallbackExecutionMode.ASYNC.value,	1✔
261	choices=[v.value for v in CallbackExecutionMode.__members__.values()])
262	state = DynamicField(default=0)	1✔
263	is_valid = BooleanField(default=True)	1✔
264	meta = {"indexes": [{"fields": ["bot", "identifier"]}]}	1✔
265
266	@staticmethod	1✔
267	def create_entry(name: str, callback_config_name: str, bot: str, sender_id: str, channel: str, metadata: dict, **kwargs):	1✔
268	check_nonempty_string(name)	1✔
269	check_nonempty_string(callback_config_name)	1✔
270	check_nonempty_string(bot)	1✔
271	check_nonempty_string(sender_id)	1✔
272	check_nonempty_string(channel)	1✔
273	identifier = f"{uuid7().hex}"	1✔
274	base_url = Utility.environment['async_callback_action']['url']	1✔
275	auth_token, is_standalone = CallbackConfig.get_auth_token(bot, callback_config_name)	1✔
276	callback_url = f"{base_url}/"	1✔
277	if is_standalone:	1✔
278	callback_url += f"s/{auth_token}"	1✔
279	else:
280	callback_url += f"d/{identifier}/{auth_token}"	1✔
281
282	record = CallbackData(action_name=name,	1✔
283	callback_name=callback_config_name,
284	bot=bot,
285	sender_id=sender_id,
286	channel=channel,
287	metadata=metadata,
288	identifier=identifier,
289	callback_url=callback_url,
290	timestamp=time.time(),
291	is_valid=True,
292	**kwargs)
293	record.save()	1✔
294	return callback_url, identifier, is_standalone	1✔
295
296	@staticmethod	1✔
297	def get_value_from_json(json_obj: Any, path: str):	1✔
298	keys = path.split('.')	1✔
299	value = json_obj	1✔
300	try:	1✔
301	for key in keys:	1✔
302	if isinstance(value, list):	1✔
303	key = int(key)	1✔
304	value = value[key]	1✔
305	except (KeyError, IndexError, ValueError, TypeError):	1✔
306	raise CallbackException(f"Cannot find identifier at path '{path}' in request data!", 200)	1✔
307
308	return value	1✔
309
310	@staticmethod	1✔
311	def validate_entry(token: str, identifier: Optional[str] = None, request_body: Any = None) -> tuple[dict, dict]:	1✔
312	check_nonempty_string(token)	1✔
313	config_entry = CallbackConfig.verify_auth_token(token)	1✔
314
315	if config_entry.standalone:	1✔
316	if not request_body:	1✔
317	raise AppException("Request data is required for standalone callbacks!")	×
318	identifier = CallbackData.get_value_from_json(request_body, config_entry.standalone_id_path)	1✔
319
320	record = CallbackData.objects(bot=config_entry.bot, identifier=identifier).first()	1✔
321	if not record:	1✔
322	raise AppException("Callback Record does not exist, invalid identifier!")	1✔
323	if not record.is_valid:	1✔
324	raise AppException("Callback has been invalidated!")	×
325	if config_entry.expire_in > 0:	1✔
326	exp_time = record.timestamp + config_entry.expire_in	×
327	if exp_time < time.time():	×
328	raise AppException("Callback time-limit expired")	×
329	entry_dict = record.to_mongo().to_dict()	1✔
330	entry_dict.pop('_id')	1✔
331	entry_dict.pop('timestamp')	1✔
332	callback_dict = config_entry.to_mongo().to_dict()	1✔
333	return entry_dict, callback_dict	1✔
334
335	@staticmethod	1✔
336	def update_state(bot: str, identifier: str, state: dict, invalidate: bool):	1✔
337	record = CallbackData.objects(bot=bot, identifier=identifier).first()	1✔
338	if not record:	1✔
UNCOV 339	raise AppException("Callback Record does not exist, invalid identifier!")	×
340	record.state = state	1✔
341	record.is_valid = not invalidate	1✔
342	record.save()	1✔
343	return record.to_mongo().to_dict()	1✔
344
345
346	@push_notification.apply	1✔
347	class CallbackLog(Document):	1✔
348	"""
349	this represents the record of actual execution record of callback after the callback url is triggered
350	"""
351	callback_name = StringField(required=True)	1✔
352	bot = StringField(required=True)	1✔
353	channel = StringField(default='unsupported')	1✔
354	identifier = StringField(required=True)	1✔
355	pyscript_code = StringField(required=True)	1✔
356	sender_id = StringField()	1✔
357	log = StringField()	1✔
358	timestamp = DateTimeField(default=datetime.utcnow)	1✔
359	status = StringField(default=CallbackRecordStatusType.SUCCESS.value,	1✔
360	choices=[v.value for v in CallbackRecordStatusType.__members__.values()])
361	request_data = DynamicField()	1✔
362	metadata = DynamicField()	1✔
363	callback_url = StringField(required=True)	1✔
364	callback_source = StringField()	1✔
365
366	meta = {"indexes": [{"fields": ["bot", "identifier"]}]}	1✔
367
368	@staticmethod	1✔
369	def create_success_entry(name: str,	1✔
370	bot: str,
371	channel: str,
372	identifier: str,
373	pyscript_code: str,
374	sender_id: str,
375	log: str,
376	request_data: Any,
377	metadata: dict,
378	callback_url: str,
379	callback_source: str) -> dict:
380	check_nonempty_string(name)	1✔
381	check_nonempty_string(bot)	1✔
382	check_nonempty_string(identifier)	1✔
383	check_nonempty_string(pyscript_code)	1✔
384	check_nonempty_string(callback_url)	1✔
385	record = CallbackLog(callback_name=name,	1✔
386	bot=bot,
387	channel=channel,
388	identifier=identifier,
389	pyscript_code=pyscript_code,
390	sender_id=sender_id,
391	log=log,
392	request_data=request_data,
393	metadata=metadata,
394	callback_url=callback_url,
395	callback_source=callback_source,
396	status=CallbackRecordStatusType.SUCCESS.value,
397	timestamp=datetime.utcnow())
398	record.save()	1✔
399	return record.to_mongo().to_dict()	1✔
400
401	@staticmethod	1✔
402	def create_failure_entry(name: str,	1✔
403	bot: str,
404	channel: str,
405	identifier: str,
406	pyscript_code: str,
407	sender_id: str,
408	error_log: str,
409	request_data: Any,
410	metadata: dict,
411	callback_url: str,
412	callback_source: str) -> dict:
413	check_nonempty_string(name)	1✔
414	check_nonempty_string(bot)	1✔
415	check_nonempty_string(identifier)	1✔
416	check_nonempty_string(pyscript_code)	1✔
417	check_nonempty_string(callback_url)	1✔
418	record = CallbackLog(callback_name=name,	1✔
419	bot=bot,
420	channel=channel,
421	identifier=identifier,
422	pyscript_code=pyscript_code,
423	sender_id=sender_id,
424	log=error_log,
425	request_data=request_data,
426	metadata=metadata,
427	callback_url=callback_url,
428	callback_source=callback_source,
429	status=CallbackRecordStatusType.FAILED.value,
430	timestamp=datetime.utcnow())
431	record.save()	1✔
432	return record.to_mongo().to_dict()	1✔
433
434	@staticmethod	1✔
435	def get_logs(query: dict, offset: int, limit: int) -> tuple[list[dict], int]:	1✔
436	logs = CallbackLog.objects(**query).skip(offset).limit(limit).exclude('id').order_by('-timestamp').to_json()	1✔
437	logs_dict_list = json.loads(logs)	1✔
438	for log in logs_dict_list:	1✔
439	log['timestamp'] = log['timestamp']['$date']	1✔
440	total = CallbackLog.objects(**query).count()	1✔
441	return logs_dict_list, total	1✔
442
443	class PyscriptPayload(BaseModel):	1✔
444	"""
445	Incoming JSON payload for restricted-Python execution.
446	"""
447	source_code: str	1✔
448	predefined_objects: Optional[Dict[str, Any]] = None	1✔
449
450
451	class CallbackRequest(BaseModel):	1✔
452	event_class: EventClass	1✔
453	data: dict	1✔
454	task_type: str	1✔

digiteinfotech / kairon / 30004242229

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous