1 | import hmac |
||
2 | import hashlib |
||
3 | |||
4 | from rest_framework.response import Response |
||
5 | from rest_framework.decorators import api_view |
||
6 | |||
7 | from django_th.models import TriggerService |
||
8 | from django_th.services import default_provider |
||
9 | |||
10 | from th_taiga.models import Taiga |
||
11 | |||
12 | |||
13 | def data_filter(trigger_id, **data): |
||
14 | """ |
||
15 | check if we want to track event for a given action |
||
16 | :param trigger_id: |
||
17 | :param data: |
||
18 | :return: |
||
19 | """ |
||
20 | taiga = Taiga.objects.get(trigger_id=trigger_id) |
||
21 | |||
22 | action = data.get('action') |
||
23 | type = data.get('type') |
||
24 | data = data.get('data') |
||
25 | |||
26 | if type == 'epic' and action == 'create' and taiga.notify_epic_create: |
||
27 | data['type_action'] = 'New Epic created' |
||
28 | elif type == 'epic' and action == 'change' and taiga.notify_epic_change: |
||
29 | data['type_action'] = 'Changed Epic' |
||
30 | elif type == 'epic' and action == 'delete' and taiga.notify_epic_delete: |
||
31 | data['type_action'] = 'Deleted Epic' |
||
32 | elif type == 'issue' and action == 'create' and \ |
||
33 | taiga.notify_issue_create: |
||
34 | data['type_action'] = 'New Issue created' |
||
35 | elif type == 'issue' and action == 'change' and \ |
||
36 | taiga.notify_issue_change: |
||
37 | data['type_action'] = 'Changed Issue' |
||
38 | elif type == 'issue' and action == 'delete' and \ |
||
39 | taiga.notify_issue_delete: |
||
40 | data['type_action'] = 'Deleted Issue' |
||
41 | elif type == 'userstory' and action == 'create' and \ |
||
42 | taiga.notify_userstory_create: |
||
43 | data['type_action'] = 'New Userstory created' |
||
44 | elif type == 'userstory' and action == 'change' and \ |
||
45 | taiga.notify_userstory_change: |
||
46 | data['type_action'] = 'Changed Userstory' |
||
47 | elif type == 'userstory' and action == 'delete' and \ |
||
48 | taiga.notify_userstory_delete: |
||
49 | data['type_action'] = 'Deleted Userstory' |
||
50 | elif type == 'task' and action == 'create' and \ |
||
51 | taiga.notify_task_create: |
||
52 | data['type_action'] = 'New Task created' |
||
53 | elif type == 'task' and action == 'change' and \ |
||
54 | taiga.notify_task_change: |
||
55 | data['type_action'] = 'Changed Task' |
||
56 | elif type == 'task' and action == 'delete' and \ |
||
57 | taiga.notify_task_delete: |
||
58 | data['type_action'] = 'Deleted Task' |
||
59 | elif type == 'wikipage' and action == 'create' and \ |
||
60 | taiga.notify_wikipage_create: |
||
61 | data['type_action'] = 'New Wikipage created' |
||
62 | elif type == 'wikipage' and action == 'change' and \ |
||
63 | taiga.notify_wikipage_change: |
||
64 | data['type_action'] = 'Changed Wikipage' |
||
65 | elif type == 'wikipage' and action == 'delete' and \ |
||
66 | taiga.notify_wikipage_delete: |
||
67 | data['type_action'] = 'Deleted Wikipage' |
||
68 | elif type == 'relateduserstory' and action == 'create' and \ |
||
69 | taiga.notify_relateduserstory_create: |
||
70 | data['type_action'] = 'New Related Userstory created' |
||
71 | elif type == 'relateduserstory' and action == 'delete' and \ |
||
72 | taiga.notify_relateduserstory_delete: |
||
73 | data['type_action'] = 'Deleted Related Userstory' |
||
74 | else: |
||
75 | data = [] |
||
76 | |||
77 | return data |
||
78 | |||
79 | |||
80 | View Code Duplication | def consumer(trigger_id, data): |
|
0 ignored issues
–
show
Duplication
introduced
by
Loading history...
|
|||
81 | """ |
||
82 | call the consumer and handle the data |
||
83 | :param trigger_id: |
||
84 | :param data: |
||
85 | :return: |
||
86 | """ |
||
87 | # consumer - the service which uses the data |
||
88 | default_provider.load_services() |
||
89 | service = TriggerService.objects.get(id=trigger_id) |
||
90 | |||
91 | service_consumer = default_provider.get_service( |
||
92 | str(service.consumer.name.name)) |
||
93 | kwargs = {'user': service.user} |
||
94 | |||
95 | data = data_filter(trigger_id, **data) |
||
96 | if len(data) > 0: |
||
97 | |||
98 | getattr(service_consumer, '__init__')(service.consumer.token, |
||
99 | **kwargs) |
||
100 | getattr(service_consumer, 'save_data')(service.id, **data) |
||
101 | |||
102 | |||
103 | def verify_signature(data, key, signature): |
||
104 | mac = hmac.new(key.encode("utf-8"), msg=data, digestmod=hashlib.sha1) |
||
105 | return mac.hexdigest() == signature |
||
106 | |||
107 | |||
108 | @api_view(['POST']) |
||
109 | def taiga(request, trigger_id, key): |
||
110 | signature = request.META.get('HTTP_X_TAIGA_WEBHOOK_SIGNATURE') |
||
111 | # check that the data are ok with the provided signature |
||
112 | if verify_signature(request._request.body, key, signature): |
||
113 | consumer(trigger_id, request.data) |
||
114 | return Response({"message": "Success"}) |
||
115 | else: |
||
116 | return Response({"message": "Failed!"}) |
||
117 |