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) · 3.18 KB
/
event.py
File metadata and controls
95 lines (74 loc) · 3.18 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 json
import logging
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.
"""
name: str
""" The name of the event. Could be `pull_request`, for example."""
delivery_id: str
"""The delivery ID of the event."""
signature: t.Optional[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`][github_bot_api.webhook.Webhook] / if the *webhook_secret* parameter is
passed to [`accept_event()`][github_bot_api.event.accept_event])."""
user_agent: str
"""The user agent invoking the webhook."""
payload: t.Dict[str, t.Any]
"""The event payload."""
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.
Args:
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 and not 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().
"""