forked from NiklasRosenstein/python-github-bot-api
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathevent.py
More file actions
95 lines (74 loc) · 2.91 KB
/
event.py
File metadata and controls
95 lines (74 loc) · 2.91 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
"""
Abstraction of a GitHub Webhook event.
Reference: https://docs.github.com/en/free-pro-team@latest/developers/webhooks-and-events/webhook-events-and-payloads
"""
import logging
import json
import typing as t
from dataclasses import dataclass
from .signature import check_signature
from .utils.mime import get_mime_components
logger = logging.getLogger(__name__)
@dataclass
class Event:
"""
Represents a GitHub webhook event.
"""
#: The name of the event. Could be `pull_request`, for example.
name: str
#: The delivery ID of the event.
delivery_id: str
#: The signature of the event. Will only be set if a Webhook-secret is configured on the
#: client side (e.g. in #Webhook.secret / if the *webhook_secret* parameter is passed to
#: #accept_event()).
signature: t.Optional[str]
#: The user agent invoking the webhook.
user_agent: str
#: The event payload.
payload: t.Dict[str, t.Any]
def accept_event(
headers: t.Mapping[str, str],
raw_body: bytes,
webhook_secret: t.Optional[str] = None,
) -> Event:
"""
Converts thee HTTP *headers* and the *raw_body* to an #Event object.
# Arguments
headers: The HTTP headers. Must have `X-Github-Event`, `X-Github-Delivery`, `User-Agent`, `Content-Type`.
May have `X-Hub-Signature` or `X-Hub-Signature-256`.
raw_body: The raw request body for the event. This is converted into a JSON payload.
webhook_secret: If specified, the `X-Hub-Signature` or `X-Hub-Signature-256` headers are used to verify
the signature of the payload. If not specified, the client does not validate the signature.
"""
event_name = headers.get('X-GitHub-Event')
delivery_id = headers.get('X-GitHub-Delivery')
signature_1 = headers.get('X-Hub-Signature')
signature_256 = headers.get('X-Hub-Signature-256')
user_agent = headers.get('User-Agent')
content_type = headers.get('Content-Type')
if not event_name or not delivery_id or not user_agent or not content_type:
raise InvalidRequest('missing required headers')
if webhook_secret is not None and not (signature_1 or signature_256):
raise InvalidRequest('webhook secret is configured but no signature header was received')
mime_type, parameters = get_mime_components(content_type)
if mime_type != 'application/json':
raise InvalidRequest(f'expected Content-Type: application/json, got {content_type}')
encoding = dict(parameters).get('encoding', 'UTF-8')
if webhook_secret is not None:
if signature_256:
check_signature(signature_256, raw_body, webhook_secret.encode('ascii'), algo='sha256')
elif signature_1:
check_signature(signature_1, raw_body, webhook_secret.encode('ascii'), algo='sha1')
else:
raise RuntimeError
return Event(
event_name,
delivery_id,
signature_256 or signature_1,
user_agent,
json.loads(raw_body.decode(encoding)),
)
class InvalidRequest(Exception):
"""
Raised when an invalid request is passed to #accept_event().
"""