Move configuration from config.py to config.json and add support for GitHub organization webhooks

remove _format_watch() from utils.github_webhook as the value of
watchers_count is always identical with stargazers_count
https://docs.github.com/en/developers/webhooks-and-events/webhook-events-and-payloads#watch
https://developer.github.com/changes/2012-09-05-watcher-api/

.gitignore

@@ -9,4 +9,4 @@ __pycache__/
 /.idea/
 
 # Configuration
-/config.py
+/config.json

README.md

@@ -25,24 +25,9 @@ You need a Telegram bot token, create a Telegram bot with
 1. Go to your GitHub project `Settings - Webhooks - Add webhook`, fill "Payload
    URL", "Content Type" (must be `application/json`) and "Secret". You can also
    do this after start running the project.
-2. Create a new `config.py` file based on this template. `chat_id` can be
-   user/group/channel id (integer) or username (string), just make sure the bot
-   is started or member of the chat with permission to send messages
-   ```
-   BOT_TOKEN = "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11"
-   PORT = 12345
-
-   GH_WEBHOOKS = {
-       "Codertocat/Hello-World": {
-           "chat_id": -1001234567890,
-           "secret": "FPAh9pwRHCLpRL7j"
-       },
-       "Octocoders/Hello-World": {
-           "chat_id": "@username",
-           "secret": "H4xvfPNCnUhPTERq"
-       }
-   }
-   ```
+2. Copy `config_sample.json` to `config.json` to configure it. `chat_id` can be
+   user/group/channel id (integer) or username (string), make sure the bot is
+   `/start`ed or member of the chat with permission to send messages
 3. Configure reverse proxy for this app, corresponding configuration for Nginx
    looks like this
    ```

config.py

@@ -0,0 +1,12 @@
+import json
+from os import environ
+
+if environ.get('DYNO'):
+    data = json.loads(environ.get("HOOK_CONFIG"))
+else:
+    with open("config.json") as f:
+        data = json.load(f)
+
+BOT_TOKEN = data['bot_token']
+PORT = data.get('port')
+GH_WEBHOOKS = data['gh_webhooks']

config_sample.json

@@ -0,0 +1,18 @@
+{
+  "bot_token": "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11",
+  "port": 12345,
+  "gh_webhooks": {
+    "Codertocat/Hello-World": {
+      "chat_id": -1001234567890,
+      "secret": "FPAh9pwRHCLpRL7j"
+    },
+    "Octocoders/Hello-World": {
+      "chat_id": "@username",
+      "secret": "H4xvfPNCnUhPTERq"
+    },
+    "octo-org": {
+      "chat_id": "@username",
+      "secret": "KLrYeiA3vNLPVbAv"
+    }
+  }
+}

main.py

@@ -18,6 +18,7 @@
 
 """
 import asyncio
+from typing import Union
 
 from aiohttp import web, ClientSession
 from aiohttp.web_request import Request
@@ -39,10 +40,10 @@ async def main(_):
 
 @routes.post("/")
 async def github_webhook_post_handler(request: Request) -> Response:
-    valid_github_webhook = await validate_github_webhook(request)
-    if not valid_github_webhook:
+    tg_chat_id: Union[str, int, bool] = await validate_github_webhook(request)
+    if not tg_chat_id:
         return web.Response(status=403, text="403: Forbidden")
-    tg_status = await send_to_telegram(session, request)
+    tg_status = await send_to_telegram(session, tg_chat_id, request)
     return web.Response(text=f"Send to Telegram: {tg_status}")
 
 

utils/github_webhook.py

@@ -22,6 +22,7 @@ import logging
 from hashlib import sha256
 from json.decoder import JSONDecodeError
 from typing import Optional
+from typing import Union
 
 from aiohttp.web_request import Request
 from multidict import CIMultiDictProxy
@@ -29,7 +30,7 @@ from multidict import CIMultiDictProxy
 from config import GH_WEBHOOKS
 
 
-async def validate_github_webhook(request: Request) -> bool:
+async def validate_github_webhook(request: Request) -> Union[str, int, bool]:
     try:
         headers = request.headers
         if not headers.get('User-Agent').startswith('GitHub-Hookshot'):
@@ -39,20 +40,35 @@ async def validate_github_webhook(request: Request) -> bool:
             logging.warning("Content type: not json")
             return False
         payload = await request.json()
-        repo_name = payload['repository']['full_name']
-        if repo_name not in GH_WEBHOOKS.keys():
-            logging.warning("Repository: not in configuration")
+        hook_target: Optional[dict] = await _get_hook_target(payload)
+        if not hook_target:
             return False
-        return await _verify_signature(
-            bytes(GH_WEBHOOKS[repo_name]['secret'], 'UTF-8'),
+        valid_signature = await _verify_signature(
+            bytes(hook_target['secret'], 'UTF-8'),
             headers.get('X-Hub-Signature-256').split('=')[1],
             await request.read()
         )
+        if valid_signature:
+            return hook_target['chat_id']
+        else:
+            return False
     except (JSONDecodeError, AttributeError) as error:
         logging.warning("Invalid: %s", error)
         return False
 
 
+async def _get_hook_target(payload: dict) -> Optional[dict]:
+    name = (payload.get('organization', {}).get('login')
+            or payload.get('repository', {}).get('full_name'))
+    if not name:
+        logging.warning("no repo or organization found")
+        return None
+    target = GH_WEBHOOKS.get(name, None)
+    if not target:
+        logging.warning("unknown repo or organization")
+    return target
+
+
 async def _verify_signature(secret: bytes, sig: str, msg: bytes) -> bool:
     mac = hmac.new(secret, msg=msg, digestmod=sha256)
     valid_signature = hmac.compare_digest(mac.hexdigest(), sig)
@@ -96,10 +112,11 @@ async def _format_discussion(payload: dict) -> str:
 
 async def _format_fork(payload: dict) -> str:
     forkee = payload['forkee']
-    return "\u2192 <a href=\"{url}\">{name}</a>".format(
+    text = ["\u2192 <a href=\"{url}\">{name}</a>".format(
         url=forkee['html_url'],
         name=forkee['full_name']
-    )
+    ), await _get_repo_star_and_fork(payload['repository'])]
+    return "\n".join(text)
 
 
 async def _format_issues(payload: dict) -> str:
@@ -153,26 +170,24 @@ async def _format_push(payload: dict) -> str:
 async def _format_star(payload: dict) -> str:
     time = payload['starred_at']
     text = [f"\u2192 starred at <code>{time}</code>"] if time else []
-    text.append(await _get_repo_watch_star_fork(payload['repository']))
+    text.append(await _get_repo_star_and_fork(payload['repository']))
     return "\n".join(text)
 
 
-async def _format_watch(payload: dict) -> str:
-    return await _get_repo_watch_star_fork(payload['repository'])
-
-
 async def _get_event_title(event: str, payload: dict) -> str:
-    sender = payload['sender']['login']
-    action = payload.get('action') or ''
+    summary = [payload['sender']['login']]
+    # if action := payload.get('action'): summary.append(action)
+    action = payload.get('action')
+    summary.append(action) if action else []
+    summary.append(event)
     return "<b>{name}</b> | <i>{summary}</i>".format(
         name=payload['repository']['full_name'],
-        summary=" ".join([sender, action, event])
+        summary=" ".join(summary)
     )
 
 
-async def _get_repo_watch_star_fork(repo: dict) -> str:
+async def _get_repo_star_and_fork(repo: dict) -> str:
     return '\u2192 ' + ", ".join([
-        f"<b>{repo['watchers_count']}</b> watchers",
         f"<b>{repo['stargazers_count']}</b> stargazers",
         f"<b>{repo['forks_count']}</b> forks"
     ])

utils/telegram.py

@@ -24,13 +24,14 @@ from typing import Union
 from aiohttp import ClientSession, ClientResponse
 from aiohttp.web_request import Request
 
-from config import BOT_TOKEN, GH_WEBHOOKS
+from config import BOT_TOKEN
 from utils.github_webhook import format_github_webhook
 
 
-async def send_to_telegram(session: ClientSession, request: Request) -> str:
+async def send_to_telegram(session: ClientSession,
+                           chat_id: Union[str, int],
+                           request: Request) -> str:
     message_text: Optional[str] = await format_github_webhook(request)
-    chat_id: int = await get_corresponding_chat_id(await request.json())
     if not message_text:
         tg_status = "nothing to send"
     else:
@@ -39,11 +40,6 @@ async def send_to_telegram(session: ClientSession, request: Request) -> str:
     return tg_status
 
 
-async def get_corresponding_chat_id(payload: dict) -> int:
-    repo_name = payload['repository']['full_name']
-    return GH_WEBHOOKS[repo_name]['chat_id']
-
-
 async def send_message(session: ClientSession,
                        chat_id: Union[int, str],
                        text: str) -> bool: