14878199224

Committed 07 May 2025 07:58AM UTC coverage: 90.474% (+0.1%) from 90.369%

Build # 14878199224

Build Type

Pull #1944

github

Committed by

web-flow

Commit Message

Merge 8dff314c4 into d9c61cc5f

Pull Request Pull Request #1944: Move main pyscript into callback service new1

Coverage Stats

291 of 317 new or added lines in 9 files covered. (91.8%)

26148 of 28901 relevant lines covered (90.47%)

0.9 hits per line

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

88.44

/kairon/shared/pyscript/callback_pyscript_utils.py

import pickle
from calendar import timegm
from datetime import datetime, date
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from smtplib import SMTP
from typing import Text, Dict, Callable, List
import base64
from apscheduler.triggers.date import DateTrigger
from apscheduler.util import obj_to_ref, astimezone
from pymongo import MongoClient
from tzlocal import get_localzone
from uuid6 import uuid7
from loguru import logger
from kairon import Utility
from kairon.events.executors.factory import ExecutorFactory
from kairon.exceptions import AppException
from kairon.shared.actions.data_objects import EmailActionConfig
from kairon.shared.actions.utils import ActionUtility
from bson import Binary
from types import ModuleType
from requests import Response
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.asymmetric import padding as asym_padding
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from kairon.shared.callback.data_objects import CallbackConfig, CallbackData
import json as jsond

from kairon.shared.chat.user_media import UserMedia


class CallbackScriptUtility:

    @staticmethod
    def generate_id():
        return uuid7().hex


    @staticmethod
    def datetime_to_utc_timestamp(timeval):
        """
        Converts a datetime instance to a timestamp.

        :type timeval: datetime
        :rtype: float

        """
        if timeval is not None:
            return timegm(timeval.utctimetuple()) + timeval.microsecond / 1000000


    @staticmethod
    def add_schedule_job(schedule_action: Text, date_time: datetime, data: Dict, timezone: Text, _id: Text = None,
                         bot: Text = None, kwargs=None):
        if not bot:
            raise AppException("Missing bot id")

        if not _id:
            _id = uuid7().hex

        if not data:
            data = {}

        data['bot'] = bot
        data['event'] = _id

        callback_config = CallbackConfig.get_entry(bot=bot, name=schedule_action)

        script = callback_config.get('pyscript_code')

        func = obj_to_ref(ExecutorFactory.get_executor().execute_task)

        schedule_data = {
            'source_code': script,
            'predefined_objects': data
        }

        args = (func, "scheduler_evaluator", schedule_data,)
        kwargs = {'task_type': "Callback"} if kwargs is None else {**kwargs, 'task_type': "Callback"}
        trigger = DateTrigger(run_date=date_time, timezone=timezone)

        next_run_time = trigger.get_next_fire_time(None, datetime.now(astimezone(timezone) or get_localzone()))

        job_kwargs = {
            'version': 1,
            'trigger': trigger,
            'executor': "default",
            'func': func,
            'args': tuple(args) if args is not None else (),
            'kwargs': kwargs,
            'id': _id,
            'name': "execute_task",
            'misfire_grace_time': 7200,
            'coalesce': True,
            'next_run_time': next_run_time,
            'max_instances': 1,
        }

        logger.info(job_kwargs)

        client = MongoClient(Utility.environment['database']['url'])
        events_db_name = Utility.environment["events"]["queue"]["name"]
        events_db = client.get_database(events_db_name)
        scheduler_collection = Utility.environment["events"]["scheduler"]["collection"]
        job_store_name = events_db.get_collection(scheduler_collection)
        event_server = Utility.environment['events']['server_url']

        job_store_name.insert_one({
            '_id': _id,
            'next_run_time': CallbackScriptUtility.datetime_to_utc_timestamp(next_run_time),
            'job_state': Binary(pickle.dumps(job_kwargs, pickle.HIGHEST_PROTOCOL))
        })

        http_response = ActionUtility.execute_http_request(
            f"{event_server}/api/events/dispatch/{_id}",
            "GET")

        if not http_response.get("success"):
            raise AppException(http_response)
        else:
            logger.info(http_response)

    @staticmethod
    def trigger_email(
                email: List[str],
                subject: str,
                body: str,
                smtp_url: str,
                smtp_port: int,
                sender_email: str,
                smtp_password: str,
                smtp_userid: str = None,
                tls: bool = False,
                content_type="html",
        ):
            """
            This is a sync email trigger.
            Sends an email to the mail id of the recipient

            :param smtp_userid:
            :param sender_email:
            :param tls:
            :param smtp_port:
            :param smtp_url:
            :param email: the mail id of the recipient
            :param smtp_password:
            :param subject: the subject of the mail
            :param body: the body of the mail
            :param content_type: "plain" or "html" content
            :return: None
            """
            smtp = None
            try:
                smtp = SMTP(smtp_url, port=smtp_port, timeout=10)
                smtp.connect(smtp_url, smtp_port)
                if tls:
                    smtp.starttls()
                smtp.login(smtp_userid if smtp_userid else sender_email, smtp_password)

                from_addr = sender_email
                mime_body = MIMEText(body, content_type)
                msg = MIMEMultipart("alternative")
                msg["Subject"] = subject
                msg["From"] = from_addr
                msg["To"] = ",".join(email)
                msg.attach(mime_body)

                smtp.sendmail(from_addr, email, msg.as_string())

            except Exception as e:
                print(f"Failed to send email: {e}")
                raise
            finally:
                if smtp:
                    try:
                        smtp.quit()
                    except Exception as quit_error:
                        print(f"Failed to quit SMTP connection cleanly: {quit_error}")


    @staticmethod
    def send_email(email_action: Text,
                   from_email: Text,
                   to_email: Text,
                   subject:  Text,
                   body: Text,
                   bot: Text):
        if not bot:
            raise AppException("Missing bot id")

        email_action_config = EmailActionConfig.objects(bot=bot, action_name=email_action).first()
        if not email_action_config:
            raise AppException(f"Email action '{email_action}' not configured for bot {bot}")
        action_config = email_action_config.to_mongo().to_dict()

        smtp_password = action_config.get('smtp_password').get("value")
        smtp_userid = action_config.get('smtp_userid').get("value")

        CallbackScriptUtility.trigger_email(
            email=[to_email],
            subject=subject,
            body=body,
            smtp_url=action_config['smtp_url'],
            smtp_port=action_config['smtp_port'],
            sender_email=from_email,
            smtp_password=smtp_password,
            smtp_userid=smtp_userid,
            tls=action_config['tls']
        )

    @staticmethod
    def perform_cleanup(local_vars: Dict):
        logger.info(f"local_vars: {local_vars}")
        filtered_locals = {}
        if local_vars:
            for key, value in local_vars.items():
                if not isinstance(value, Callable) and not isinstance(value, ModuleType):
                    if isinstance(value, datetime):
                        value = value.strftime("%m/%d/%Y, %H:%M:%S")
                    elif isinstance(value, date):
                        value = value.strftime("%Y-%m-%d")
                    elif isinstance(value, Response):
                        value = value.text
                    filtered_locals[key] = value
        logger.info(f"filtered_vars: {filtered_locals}")
        return filtered_locals


    @staticmethod
    def decrypt_request(request_body, private_key_pem):
        try:
            encrypted_data_b64 = request_body.get("encrypted_flow_data")
            encrypted_aes_key_b64 = request_body.get("encrypted_aes_key")
            iv_b64 = request_body.get("initial_vector")

            if not (encrypted_data_b64 and encrypted_aes_key_b64 and iv_b64):
                raise ValueError("Missing required encrypted data fields")

            # Decode base64 inputs
            encrypted_aes_key = base64.b64decode(encrypted_aes_key_b64)
            encrypted_data = base64.b64decode(encrypted_data_b64)
            iv = base64.b64decode(iv_b64)[:16]  # Ensure IV is exactly 16 bytes

            private_key = load_pem_private_key(private_key_pem.encode(), password=None)

            # Decrypt AES key using RSA and OAEP padding
            aes_key = private_key.decrypt(
                encrypted_aes_key,
                asym_padding.OAEP(
                    mgf=asym_padding.MGF1(algorithm=hashes.SHA256()),
                    algorithm=hashes.SHA256(),
                    label=None,
                ),
            )

            if len(aes_key) not in (16, 24, 32):
                raise ValueError(f"Invalid AES key size: {len(aes_key)} bytes")

            # Extract GCM tag (last 16 bytes)
            encrypted_body = encrypted_data[:-16]
            tag = encrypted_data[-16:]

            # Decrypt AES-GCM
            cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv, tag))
            decryptor = cipher.decryptor()
            decrypted_bytes = decryptor.update(encrypted_body) + decryptor.finalize()
            decrypted_data = jsond.loads(decrypted_bytes.decode("utf-8"))

            response_dict = {
                "decryptedBody": decrypted_data,
                "aesKeyBuffer": aes_key,
                "initialVectorBuffer": iv,
            }

            return response_dict

        except Exception as e:
            raise Exception(f"decryption failed-{str(e)}")


    @staticmethod
    def encrypt_response(response_body, aes_key_buffer, initial_vector_buffer):
        try:
            if aes_key_buffer is None:
                raise ValueError("AES key cannot be None")

            if initial_vector_buffer is None:
                raise ValueError("Initialization vector (IV) cannot be None")

            # Flip the IV
            flipped_iv = bytes(byte ^ 0xFF for byte in initial_vector_buffer)

            # Encrypt using AES-GCM
            encryptor = Cipher(algorithms.AES(aes_key_buffer), modes.GCM(flipped_iv)).encryptor()
            encrypted_bytes = encryptor.update(jsond.dumps(response_body).encode("utf-8")) + encryptor.finalize()
            encrypted_data_with_tag = encrypted_bytes + encryptor.tag

            # Encode result as base64
            encoded_data = base64.b64encode(encrypted_data_with_tag).decode("utf-8")
            return encoded_data
        except Exception as e:
            raise Exception(f"encryption failed-{str(e)}")


    @staticmethod
    def create_callback(callback_name: str, metadata: dict, bot: str, sender_id: str, channel: str,
                        name: str = 'callback_pyscript'):
        callback_url, identifier, standalone = CallbackData.create_entry(
            name=name,
            callback_config_name=callback_name,
            bot=bot,
            sender_id=sender_id,
            channel=channel,
            metadata=metadata,
        )
        if standalone:
            return identifier
        else:
            return callback_url

    @staticmethod
    def save_as_pdf(text: str, bot: str, sender_id:str):
        try:
            _, media_id = UserMedia.save_markdown_as_pdf(
                bot=bot,
                sender_id=sender_id,
                text=text,
                filepath="report.pdf"
            )
            return media_id
        except Exception as e:
            raise Exception(f"encryption failed-{str(e)}")

1	import pickle	1✔
2	from calendar import timegm	1✔
3	from datetime import datetime, date	1✔
4	from email.mime.multipart import MIMEMultipart	1✔
5	from email.mime.text import MIMEText	1✔
6	from smtplib import SMTP	1✔
7	from typing import Text, Dict, Callable, List	1✔
8	import base64	1✔
9	from apscheduler.triggers.date import DateTrigger	1✔
10	from apscheduler.util import obj_to_ref, astimezone	1✔
11	from pymongo import MongoClient	1✔
12	from tzlocal import get_localzone	1✔
13	from uuid6 import uuid7	1✔
14	from loguru import logger	1✔
15	from kairon import Utility	1✔
16	from kairon.events.executors.factory import ExecutorFactory	1✔
17	from kairon.exceptions import AppException	1✔
18	from kairon.shared.actions.data_objects import EmailActionConfig	1✔
19	from kairon.shared.actions.utils import ActionUtility	1✔
20	from bson import Binary	1✔
21	from types import ModuleType	1✔
22	from requests import Response	1✔
23	from cryptography.hazmat.primitives import hashes	1✔
24	from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes	1✔
25	from cryptography.hazmat.primitives.asymmetric import padding as asym_padding	1✔
26	from cryptography.hazmat.primitives.serialization import load_pem_private_key	1✔
27	from kairon.shared.callback.data_objects import CallbackConfig, CallbackData	1✔
28	import json as jsond	1✔
29
30	from kairon.shared.chat.user_media import UserMedia	1✔
31
32
33	class CallbackScriptUtility:	1✔
34
35	@staticmethod	1✔
36	def generate_id():	1✔
37	return uuid7().hex	1✔
38
39
40	@staticmethod	1✔
41	def datetime_to_utc_timestamp(timeval):	1✔
42	"""
43	Converts a datetime instance to a timestamp.
44
45	:type timeval: datetime
46	:rtype: float
47
48	"""
49	if timeval is not None:	1✔
50	return timegm(timeval.utctimetuple()) + timeval.microsecond / 1000000	1✔
51
52
53	@staticmethod	1✔
54	def add_schedule_job(schedule_action: Text, date_time: datetime, data: Dict, timezone: Text, _id: Text = None,	1✔
55	bot: Text = None, kwargs=None):
56	if not bot:	1✔
57	raise AppException("Missing bot id")	1✔
58
59	if not _id:	1✔
60	_id = uuid7().hex	1✔
61
62	if not data:	1✔
63	data = {}	1✔
64
65	data['bot'] = bot	1✔
66	data['event'] = _id	1✔
67
68	callback_config = CallbackConfig.get_entry(bot=bot, name=schedule_action)	1✔
69
70	script = callback_config.get('pyscript_code')	1✔
71
72	func = obj_to_ref(ExecutorFactory.get_executor().execute_task)	1✔
73
74	schedule_data = {	1✔
75	'source_code': script,
76	'predefined_objects': data
77	}
78
79	args = (func, "scheduler_evaluator", schedule_data,)	1✔
80	kwargs = {'task_type': "Callback"} if kwargs is None else {**kwargs, 'task_type': "Callback"}	1✔
81	trigger = DateTrigger(run_date=date_time, timezone=timezone)	1✔
82
83	next_run_time = trigger.get_next_fire_time(None, datetime.now(astimezone(timezone) or get_localzone()))	1✔
84
85	job_kwargs = {	1✔
86	'version': 1,
87	'trigger': trigger,
88	'executor': "default",
89	'func': func,
90	'args': tuple(args) if args is not None else (),
91	'kwargs': kwargs,
92	'id': _id,
93	'name': "execute_task",
94	'misfire_grace_time': 7200,
95	'coalesce': True,
96	'next_run_time': next_run_time,
97	'max_instances': 1,
98	}
99
100	logger.info(job_kwargs)	1✔
101
102	client = MongoClient(Utility.environment['database']['url'])	1✔
103	events_db_name = Utility.environment["events"]["queue"]["name"]	1✔
104	events_db = client.get_database(events_db_name)	1✔
105	scheduler_collection = Utility.environment["events"]["scheduler"]["collection"]	1✔
106	job_store_name = events_db.get_collection(scheduler_collection)	1✔
107	event_server = Utility.environment['events']['server_url']	1✔
108
109	job_store_name.insert_one({	1✔
110	'_id': _id,
111	'next_run_time': CallbackScriptUtility.datetime_to_utc_timestamp(next_run_time),
112	'job_state': Binary(pickle.dumps(job_kwargs, pickle.HIGHEST_PROTOCOL))
113	})
114
115	http_response = ActionUtility.execute_http_request(	1✔
116	f"{event_server}/api/events/dispatch/{_id}",
117	"GET")
118
119	if not http_response.get("success"):	1✔
120	raise AppException(http_response)	1✔
121	else:
122	logger.info(http_response)	1✔
123
124	@staticmethod	1✔
125	def trigger_email(	1✔
126	email: List[str],
127	subject: str,
128	body: str,
129	smtp_url: str,
130	smtp_port: int,
131	sender_email: str,
132	smtp_password: str,
133	smtp_userid: str = None,
134	tls: bool = False,
135	content_type="html",
136	):
137	"""
138	This is a sync email trigger.
139	Sends an email to the mail id of the recipient
140
141	:param smtp_userid:
142	:param sender_email:
143	:param tls:
144	:param smtp_port:
145	:param smtp_url:
146	:param email: the mail id of the recipient
147	:param smtp_password:
148	:param subject: the subject of the mail
149	:param body: the body of the mail
150	:param content_type: "plain" or "html" content
151	:return: None
152	"""
153	smtp = None	1✔
154	try:	1✔
155	smtp = SMTP(smtp_url, port=smtp_port, timeout=10)	1✔
156	smtp.connect(smtp_url, smtp_port)	1✔
157	if tls:	1✔
158	smtp.starttls()	1✔
159	smtp.login(smtp_userid if smtp_userid else sender_email, smtp_password)	1✔
160
161	from_addr = sender_email	1✔
162	mime_body = MIMEText(body, content_type)	1✔
163	msg = MIMEMultipart("alternative")	1✔
164	msg["Subject"] = subject	1✔
165	msg["From"] = from_addr	1✔
166	msg["To"] = ",".join(email)	1✔
167	msg.attach(mime_body)	1✔
168
169	smtp.sendmail(from_addr, email, msg.as_string())	1✔
170
NEW 171	except Exception as e:	×
NEW 172	print(f"Failed to send email: {e}")	×
NEW 173	raise	×
174	finally:
175	if smtp:	1✔
176	try:	1✔
177	smtp.quit()	1✔
NEW 178	except Exception as quit_error:	×
NEW 179	print(f"Failed to quit SMTP connection cleanly: {quit_error}")	×
180
181
182	@staticmethod	1✔
183	def send_email(email_action: Text,	1✔
184	from_email: Text,
185	to_email: Text,
186	subject: Text,
187	body: Text,
188	bot: Text):
189	if not bot:	1✔
190	raise AppException("Missing bot id")	1✔
191
192	email_action_config = EmailActionConfig.objects(bot=bot, action_name=email_action).first()	1✔
193	if not email_action_config:	1✔
194	raise AppException(f"Email action '{email_action}' not configured for bot {bot}")	1✔
195	action_config = email_action_config.to_mongo().to_dict()	1✔
196
197	smtp_password = action_config.get('smtp_password').get("value")	1✔
198	smtp_userid = action_config.get('smtp_userid').get("value")	1✔
199
200	CallbackScriptUtility.trigger_email(	1✔
201	email=[to_email],
202	subject=subject,
203	body=body,
204	smtp_url=action_config['smtp_url'],
205	smtp_port=action_config['smtp_port'],
206	sender_email=from_email,
207	smtp_password=smtp_password,
208	smtp_userid=smtp_userid,
209	tls=action_config['tls']
210	)
211
212	@staticmethod	1✔
213	def perform_cleanup(local_vars: Dict):	1✔
NEW 214	logger.info(f"local_vars: {local_vars}")	×
NEW 215	filtered_locals = {}	×
NEW 216	if local_vars:	×
NEW 217	for key, value in local_vars.items():	×
NEW 218	if not isinstance(value, Callable) and not isinstance(value, ModuleType):	×
NEW 219	if isinstance(value, datetime):	×
NEW 220	value = value.strftime("%m/%d/%Y, %H:%M:%S")	×
NEW 221	elif isinstance(value, date):	×
NEW 222	value = value.strftime("%Y-%m-%d")	×
NEW 223	elif isinstance(value, Response):	×
NEW 224	value = value.text	×
NEW 225	filtered_locals[key] = value	×
NEW 226	logger.info(f"filtered_vars: {filtered_locals}")	×
NEW 227	return filtered_locals	×
228
229
230	@staticmethod	1✔
231	def decrypt_request(request_body, private_key_pem):	1✔
232	try:	1✔
233	encrypted_data_b64 = request_body.get("encrypted_flow_data")	1✔
234	encrypted_aes_key_b64 = request_body.get("encrypted_aes_key")	1✔
235	iv_b64 = request_body.get("initial_vector")	1✔
236
237	if not (encrypted_data_b64 and encrypted_aes_key_b64 and iv_b64):	1✔
238	raise ValueError("Missing required encrypted data fields")	1✔
239
240	# Decode base64 inputs
241	encrypted_aes_key = base64.b64decode(encrypted_aes_key_b64)	1✔
242	encrypted_data = base64.b64decode(encrypted_data_b64)	1✔
243	iv = base64.b64decode(iv_b64)[:16] # Ensure IV is exactly 16 bytes	1✔
244
245	private_key = load_pem_private_key(private_key_pem.encode(), password=None)	1✔
246
247	# Decrypt AES key using RSA and OAEP padding
248	aes_key = private_key.decrypt(	1✔
249	encrypted_aes_key,
250	asym_padding.OAEP(
251	mgf=asym_padding.MGF1(algorithm=hashes.SHA256()),
252	algorithm=hashes.SHA256(),
253	label=None,
254	),
255	)
256
257	if len(aes_key) not in (16, 24, 32):	1✔
NEW 258	raise ValueError(f"Invalid AES key size: {len(aes_key)} bytes")	×
259
260	# Extract GCM tag (last 16 bytes)
261	encrypted_body = encrypted_data[:-16]	1✔
262	tag = encrypted_data[-16:]	1✔
263
264	# Decrypt AES-GCM
265	cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv, tag))	1✔
266	decryptor = cipher.decryptor()	1✔
267	decrypted_bytes = decryptor.update(encrypted_body) + decryptor.finalize()	1✔
268	decrypted_data = jsond.loads(decrypted_bytes.decode("utf-8"))	1✔
269
270	response_dict = {	1✔
271	"decryptedBody": decrypted_data,
272	"aesKeyBuffer": aes_key,
273	"initialVectorBuffer": iv,
274	}
275
276	return response_dict	1✔
277
278	except Exception as e:	1✔
279	raise Exception(f"decryption failed-{str(e)}")	1✔
280
281
282	@staticmethod	1✔
283	def encrypt_response(response_body, aes_key_buffer, initial_vector_buffer):	1✔
284	try:	1✔
285	if aes_key_buffer is None:	1✔
286	raise ValueError("AES key cannot be None")	1✔
287
288	if initial_vector_buffer is None:	1✔
289	raise ValueError("Initialization vector (IV) cannot be None")	1✔
290
291	# Flip the IV
292	flipped_iv = bytes(byte ^ 0xFF for byte in initial_vector_buffer)	1✔
293
294	# Encrypt using AES-GCM
295	encryptor = Cipher(algorithms.AES(aes_key_buffer), modes.GCM(flipped_iv)).encryptor()	1✔
296	encrypted_bytes = encryptor.update(jsond.dumps(response_body).encode("utf-8")) + encryptor.finalize()	1✔
297	encrypted_data_with_tag = encrypted_bytes + encryptor.tag	1✔
298
299	# Encode result as base64
300	encoded_data = base64.b64encode(encrypted_data_with_tag).decode("utf-8")	1✔
301	return encoded_data	1✔
302	except Exception as e:	1✔
303	raise Exception(f"encryption failed-{str(e)}")	1✔
304
305
306	@staticmethod	1✔
307	def create_callback(callback_name: str, metadata: dict, bot: str, sender_id: str, channel: str,	1✔
308	name: str = 'callback_pyscript'):
309	callback_url, identifier, standalone = CallbackData.create_entry(	1✔
310	name=name,
311	callback_config_name=callback_name,
312	bot=bot,
313	sender_id=sender_id,
314	channel=channel,
315	metadata=metadata,
316	)
317	if standalone:	1✔
318	return identifier	1✔
319	else:
320	return callback_url	1✔
321
322	@staticmethod	1✔
323	def save_as_pdf(text: str, bot: str, sender_id:str):	1✔
324	try:	1✔
325	_, media_id = UserMedia.save_markdown_as_pdf(	1✔
326	bot=bot,
327	sender_id=sender_id,
328	text=text,
329	filepath="report.pdf"
330	)
331	return media_id	1✔
332	except Exception as e:	1✔
333	raise Exception(f"encryption failed-{str(e)}")	1✔

digiteinfotech / kairon / 14878199224

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