(request: Request)
| 32 | |
| 33 | |
| 34 | async def validate_github_webhook(request: Request) -> Union[str, int, bool]: |
| 35 | try: |
| 36 | headers = request.headers |
| 37 | if not headers.get('User-Agent').startswith('GitHub-Hookshot'): |
| 38 | logging.warning("User agent: not from GitHub") |
| 39 | return False |
| 40 | if headers.get('Content-Type') != 'application/json': |
| 41 | logging.warning("Content type: not json") |
| 42 | return False |
| 43 | payload = await request.json() |
| 44 | hook_target: Optional[dict] = await _get_hook_target(payload) |
| 45 | if not hook_target: |
| 46 | return False |
| 47 | valid_signature = await _verify_signature( |
| 48 | bytes(hook_target['secret'], 'UTF-8'), |
| 49 | headers.get('X-Hub-Signature-256').split('=')[1], |
| 50 | await request.read() |
| 51 | ) |
| 52 | if valid_signature: |
| 53 | return hook_target['chat_id'] |
| 54 | else: |
| 55 | return False |
| 56 | except (JSONDecodeError, AttributeError) as error: |
| 57 | logging.warning("Invalid: %s", error) |
| 58 | return False |
| 59 | |
| 60 | |
| 61 | async def _get_hook_target(payload: dict) -> Optional[dict]: |
no test coverage detected