Coverage for documents/workflows/webhooks.py: 27%

1import logging 

2 

3import httpx 

4from celery import shared_task 

5from django.conf import settings 

6 

7from paperless.network import PinnedHostHTTPTransport 

8from paperless.network import validate_outbound_http_url 

9 

10logger = logging.getLogger("paperless.workflows.webhooks") 

11 

12 

13@shared_task( 

14 retry_backoff=True, 

15 autoretry_for=(httpx.HTTPStatusError,), 

16 max_retries=3, 

17 throws=(httpx.HTTPError,), 

18) 

19def send_webhook( 

20 url: str, 

21 data: str | dict, 

22 headers: dict, 

23 files: dict, 

24 *, 

25 as_json: bool = False, 

26): 

27 try: 

28 validate_outbound_http_url( 

29 url, 

30 allowed_schemes=settings.WEBHOOKS_ALLOWED_SCHEMES, 

31 allowed_ports=settings.WEBHOOKS_ALLOWED_PORTS, 

32 # Internal-address checks happen in transport to preserve ConnectError behavior. 

33 allow_internal=True, 

34 ) 

35 except ValueError as e: 

36 logger.warning("Webhook blocked: %s", e) 

37 raise 

38 

39 transport = PinnedHostHTTPTransport( 

40 allow_internal=settings.WEBHOOKS_ALLOW_INTERNAL_REQUESTS, 

41 ) 

42 

43 try: 

44 post_args = { 

45 "url": url, 

46 "headers": { 

47 k: v for k, v in (headers or {}).items() if k.lower() != "host" 

48 }, 

49 "files": files or None, 

50 } 

51 if as_json: 

52 post_args["json"] = data 

53 elif isinstance(data, dict): 

54 post_args["data"] = data 

55 else: 

56 post_args["content"] = data 

57 

58 with httpx.Client( 

59 transport=transport, 

60 timeout=5.0, 

61 follow_redirects=False, 

62 ) as client: 

63 client.post( 

64 **post_args, 

65 ).raise_for_status() 

66 logger.info( 

67 f"Webhook sent to {url}", 

68 ) 

69 except Exception as e: 

70 logger.error( 

71 f"Failed attempt sending webhook to {url}: {e}", 

72 ) 

73 raise e 

74 finally: 

75 transport.close()