Coverage for open_webui/utils/webhook.py: 13%

1import asyncio 

2import logging 

3 

4from open_webui.config import WEBUI_FAVICON_URL 

5from open_webui.env import ( 

6 AIOHTTP_CLIENT_ALLOW_REDIRECTS, 

7 AIOHTTP_CLIENT_SESSION_SSL, 

8 VERSION, 

9) 

10from open_webui.retrieval.web.utils import get_ssrf_safe_session, validate_url 

11from open_webui.utils.json_codec import JSONCodec 

12 

13log = logging.getLogger(__name__) 

14 

15 

16# Let this message reach those for whom it was written, and 

17# may no network partition deny the word its destination. 

18def _event_text(message: str, description: str | None = None, event_data: dict | None = None) -> str: 

19 lines = [message] 

20 if description and description != message: 

21 lines.append(description) 

22 

23 event_name = (event_data or {}).get('event') 

24 if event_name: 

25 lines.append(f'Event: {event_name}') 

26 

27 return '\n'.join(lines) 

28 

29 

30async def post_webhook(name: str, url: str, message: str, event_data: dict, description: str | None = None) -> bool: 

31 try: 

32 log.debug('post_webhook: %s, %s, %s', url, message, event_data) 

33 # Block private-IP / loopback / cloud-metadata targets — the URL is 

34 # caller-controlled (user notification settings under 

35 # ENABLE_USER_WEBHOOKS, automation notification triggers). 

36 await asyncio.to_thread(validate_url, url) 

37 except Exception as e: 

38 log.warning('Webhook skipped, URL invalid or not publicly resolvable: %s', e) 

39 return False 

40 

41 try: 

42 payload = {} 

43 # Slack and Google Chat Webhooks 

44 if 'https://hooks.slack.com' in url or 'https://chat.googleapis.com' in url: 

45 payload['text'] = _event_text(message, description, event_data) 

46 # Discord Webhooks 

47 elif 'https://discord.com/api/webhooks' in url: 

48 content = _event_text(message, description, event_data) 

49 payload['content'] = content if len(content) < 2000 else f'{content[: 2000 - 20]}... (truncated)' 

50 # Microsoft Teams Webhooks 

51 elif 'webhook.office.com' in url: 

52 action = event_data.get('action', 'undefined') 

53 user_data = event_data.get('user') or event_data.get('actor') or {} 

54 if isinstance(user_data, dict): 

55 user_dict = user_data 

56 else: 

57 user_dict = JSONCodec.loads(user_data) 

58 facts = [{'name': key, 'value': value} for key, value in user_dict.items()] 

59 if event_data.get('event'): 

60 facts.insert(0, {'name': 'event', 'value': event_data.get('event')}) 

61 if description: 

62 facts.insert(0, {'name': 'description', 'value': description}) 

63 payload = { 

64 '@type': 'MessageCard', 

65 '@context': 'http://schema.org/extensions', 

66 'themeColor': '0076D7', 

67 'summary': message, 

68 'sections': [ 

69 { 

70 'activityTitle': message, 

71 'activitySubtitle': f'{name} ({VERSION}) - {action}', 

72 # LICENSE covers this Open WebUI webhook logo. 

73 # Do not alter, remove, obscure, or replace it except as LICENSE permits: 

74 # https://docs.openwebui.com/license. 

75 'activityImage': WEBUI_FAVICON_URL, 

76 'text': description, 

77 'facts': facts, 

78 'markdown': True, 

79 } 

80 ], 

81 } 

82 # Default Payload 

83 else: 

84 payload = event_data 

85 

86 log.debug('payload: %s', payload) 

87 async with get_ssrf_safe_session() as session: 

88 async with session.post( 

89 url, 

90 json=payload, 

91 ssl=AIOHTTP_CLIENT_SESSION_SSL, 

92 allow_redirects=AIOHTTP_CLIENT_ALLOW_REDIRECTS, 

93 ) as r: 

94 r_text = await r.text() 

95 r.raise_for_status() 

96 log.debug('r.text: %s', r_text) 

97 

98 return True 

99 except Exception as e: 

100 log.exception(e) 

101 return False