Compare commits
2 Commits
b2180a4a34
...
hermes/her
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
27c8cb3597 | ||
|
|
4a33d6924e |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -2,4 +2,3 @@ __pycache__/
|
|||||||
*.pyc
|
*.pyc
|
||||||
*.pyo
|
*.pyo
|
||||||
.DS_Store
|
.DS_Store
|
||||||
.worktrees/
|
|
||||||
|
|||||||
@@ -4,13 +4,11 @@ Polymarket Event Browser
|
|||||||
Browse tradeable Polymarket events by game category.
|
Browse tradeable Polymarket events by game category.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import html
|
import subprocess
|
||||||
import json
|
import json
|
||||||
import time
|
import time
|
||||||
import argparse
|
import argparse
|
||||||
from datetime import datetime, timezone, timedelta
|
from datetime import datetime, timezone, timedelta
|
||||||
from urllib.parse import urlencode
|
|
||||||
from urllib.request import urlopen, Request
|
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# CONFIG
|
# CONFIG
|
||||||
@@ -578,44 +576,14 @@ def print_detail(e, detail):
|
|||||||
# TELEGRAM
|
# TELEGRAM
|
||||||
# ============================================================
|
# ============================================================
|
||||||
|
|
||||||
def escape_html(text):
|
|
||||||
"""Escape HTML-sensitive characters for Telegram parse_mode=HTML."""
|
|
||||||
return (text
|
|
||||||
.replace("&", "&")
|
|
||||||
.replace("<", "<")
|
|
||||||
.replace(">", ">")
|
|
||||||
.replace('"', """))
|
|
||||||
|
|
||||||
|
|
||||||
def send_telegram_message(bot_token, chat_id, text, timeout=10):
|
|
||||||
"""Send a message via Telegram bot API. Returns the message ID on success.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
RuntimeError: If the Telegram API returns an error (e.g. invalid token, rate limit).
|
|
||||||
URLError/HTTPError: On network or HTTP-level failures.
|
|
||||||
"""
|
|
||||||
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
|
|
||||||
data = urlencode({
|
|
||||||
"chat_id": chat_id,
|
|
||||||
"text": text,
|
|
||||||
"parse_mode": "HTML",
|
|
||||||
"disable_web_page_preview": "true",
|
|
||||||
}).encode("utf-8")
|
|
||||||
req = Request(url, data=data, method="POST")
|
|
||||||
with urlopen(req, timeout=timeout) as resp:
|
|
||||||
result = json.loads(resp.read())
|
|
||||||
if not result.get("ok"):
|
|
||||||
raise RuntimeError(f"Telegram API error: {result.get('description')}")
|
|
||||||
return result["result"]["message_id"]
|
|
||||||
|
|
||||||
|
|
||||||
def send_to_telegram(match_events, non_match_events, category, matches_only=False, non_matches_only=False):
|
def send_to_telegram(match_events, non_match_events, category, matches_only=False, non_matches_only=False):
|
||||||
"""Send browse results to Telegram. Reads TELEGRAM_BOT_TOKEN and CHAT_ID from environment."""
|
"""Send browse results to Telegram. Reads BOT_TOKEN and CHAT_ID from environment."""
|
||||||
import os
|
import os
|
||||||
bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
|
bot_token = os.environ.get("BOT_TOKEN")
|
||||||
chat_id = os.environ.get("CHAT_ID")
|
chat_id = os.environ.get("CHAT_ID")
|
||||||
if not bot_token or not chat_id:
|
if not bot_token or not chat_id:
|
||||||
raise RuntimeError("TELEGRAM_BOT_TOKEN or CHAT_ID not set in environment")
|
print("WARNING: BOT_TOKEN or CHAT_ID not set in environment. Skipping Telegram send.")
|
||||||
|
return
|
||||||
|
|
||||||
from datetime import datetime, timezone, timedelta
|
from datetime import datetime, timezone, timedelta
|
||||||
now_utc = datetime.now(timezone.utc)
|
now_utc = datetime.now(timezone.utc)
|
||||||
@@ -628,8 +596,19 @@ def send_to_telegram(match_events, non_match_events, category, matches_only=Fals
|
|||||||
show_non_matches = (not matches_only and not non_matches_only) or non_matches_only
|
show_non_matches = (not matches_only and not non_matches_only) or non_matches_only
|
||||||
|
|
||||||
def send(text):
|
def send(text):
|
||||||
msg_id = send_telegram_message(bot_token, chat_id, text)
|
result = subprocess.run(
|
||||||
print(f" Sent msg {msg_id}")
|
["curl", "-s", f"https://api.telegram.org/bot{bot_token}/sendMessage",
|
||||||
|
"-d", f"chat_id={chat_id}",
|
||||||
|
"-d", f"text={text}",
|
||||||
|
"-d", "parse_mode=HTML",
|
||||||
|
"-d", "disable_web_page_preview=true"],
|
||||||
|
capture_output=True
|
||||||
|
)
|
||||||
|
resp = json.loads(result.stdout.decode())
|
||||||
|
if resp.get("ok"):
|
||||||
|
print(f" Sent msg {resp['result']['message_id']}")
|
||||||
|
else:
|
||||||
|
print(f" Error: {resp.get('description')}")
|
||||||
|
|
||||||
# Build sections
|
# Build sections
|
||||||
lines = [f"<b>{category.upper()}</b> | {header_date}"]
|
lines = [f"<b>{category.upper()}</b> | {header_date}"]
|
||||||
@@ -655,7 +634,7 @@ def send_to_telegram(match_events, non_match_events, category, matches_only=Fals
|
|||||||
odds_b = format_odds(float(prices[1])) if len(prices) > 1 else "?"
|
odds_b = format_odds(float(prices[1])) if len(prices) > 1 else "?"
|
||||||
tournament = get_tournament(title)
|
tournament = get_tournament(title)
|
||||||
title_clean = title.split(" - ")[0].strip() if " - " in title else title
|
title_clean = title.split(" - ")[0].strip() if " - " in title else title
|
||||||
lines.append(f"<b>{i}.</b> <a href=\"{url}\">{escape_html(title_clean)}</a>")
|
lines.append(f"<b>{i}.</b> <a href=\"{url}\">{title_clean}</a>")
|
||||||
lines.append(f" {start_time_wib} | {rel_time}")
|
lines.append(f" {start_time_wib} | {rel_time}")
|
||||||
lines.append(f" Vol: ${vol:,.0f}")
|
lines.append(f" Vol: ${vol:,.0f}")
|
||||||
if tournament:
|
if tournament:
|
||||||
@@ -676,7 +655,7 @@ def send_to_telegram(match_events, non_match_events, category, matches_only=Fals
|
|||||||
start_time_wib, rel_time = get_start_time_wib(e)
|
start_time_wib, rel_time = get_start_time_wib(e)
|
||||||
total_vol = sum(float(m.get("volume", 0)) for m in e.get("markets", []))
|
total_vol = sum(float(m.get("volume", 0)) for m in e.get("markets", []))
|
||||||
market_count = len(e.get("markets", []))
|
market_count = len(e.get("markets", []))
|
||||||
lines.append(f"<b>{i}.</b> <a href=\"{url}\">{escape_html(title)}</a>")
|
lines.append(f"<b>{i}.</b> <a href=\"{url}\">{title}</a>")
|
||||||
lines.append(f" {start_time_wib} | {rel_time}")
|
lines.append(f" {start_time_wib} | {rel_time}")
|
||||||
lines.append(f" Markets: {market_count} | Total Vol: ${total_vol:,.0f}")
|
lines.append(f" Markets: {market_count} | Total Vol: ${total_vol:,.0f}")
|
||||||
lines.append("")
|
lines.append("")
|
||||||
@@ -758,7 +737,7 @@ def main():
|
|||||||
parser.add_argument("--raw", action="store_true",
|
parser.add_argument("--raw", action="store_true",
|
||||||
help="Show all events without tradeable filter (for debugging).")
|
help="Show all events without tradeable filter (for debugging).")
|
||||||
parser.add_argument("--telegram", action="store_true",
|
parser.add_argument("--telegram", action="store_true",
|
||||||
help="Send results to Telegram (TELEGRAM_BOT_TOKEN and CHAT_ID must be set in environment).")
|
help="Send results to Telegram (BOT_TOKEN and CHAT_ID must be set in environment).")
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
if args.list_categories:
|
if args.list_categories:
|
||||||
|
|||||||
@@ -1 +0,0 @@
|
|||||||
# Tests package
|
|
||||||
@@ -1,205 +0,0 @@
|
|||||||
"""
|
|
||||||
Unit tests for browse.py Telegram functions.
|
|
||||||
|
|
||||||
Run with: python -m pytest tests/test_browse.py -v
|
|
||||||
"""
|
|
||||||
|
|
||||||
import unittest
|
|
||||||
from unittest.mock import patch, MagicMock
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
|
|
||||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'scripts'))
|
|
||||||
from browse import send_telegram_message
|
|
||||||
|
|
||||||
|
|
||||||
class TestSendTelegramMessage(unittest.TestCase):
|
|
||||||
"""Tests for the module-level send_telegram_message function."""
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_success(self, mock_urlopen):
|
|
||||||
"""send_telegram_message returns message_id on success."""
|
|
||||||
mock_resp = MagicMock()
|
|
||||||
mock_resp.read.return_value = b'{"ok": true, "result": {"message_id": 123}}'
|
|
||||||
mock_urlopen.return_value.__enter__.return_value = mock_resp
|
|
||||||
|
|
||||||
result = send_telegram_message("test_token", "test_chat", "hello world")
|
|
||||||
|
|
||||||
self.assertEqual(result, 123)
|
|
||||||
mock_urlopen.assert_called_once()
|
|
||||||
call_args = mock_urlopen.call_args
|
|
||||||
req = call_args[0][0]
|
|
||||||
self.assertEqual(req.full_url, "https://api.telegram.org/bottest_token/sendMessage")
|
|
||||||
self.assertEqual(req.method, "POST")
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_api_error_raises_runtime_error(self, mock_urlopen):
|
|
||||||
"""send_telegram_message raises RuntimeError when Telegram API returns ok=false."""
|
|
||||||
mock_resp = MagicMock()
|
|
||||||
mock_resp.read.return_value = b'{"ok": false, "description": "Forbidden"}'
|
|
||||||
mock_urlopen.return_value.__enter__.return_value = mock_resp
|
|
||||||
|
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
|
||||||
send_telegram_message("test_token", "test_chat", "hello")
|
|
||||||
self.assertIn("Telegram API error: Forbidden", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_invalid_token_raises_http_error(self, mock_urlopen):
|
|
||||||
"""send_telegram_message raises HTTPError on invalid token (404)."""
|
|
||||||
from urllib.error import HTTPError
|
|
||||||
mock_urlopen.side_effect = HTTPError(
|
|
||||||
url="https://api.telegram.org/botINVALID/sendMessage",
|
|
||||||
code=404,
|
|
||||||
msg="Not Found",
|
|
||||||
hdrs={},
|
|
||||||
fp=None
|
|
||||||
)
|
|
||||||
|
|
||||||
with self.assertRaises(HTTPError) as ctx:
|
|
||||||
send_telegram_message("INVALID", "test_chat", "hello")
|
|
||||||
self.assertEqual(ctx.exception.code, 404)
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_rate_limit_raises_http_error(self, mock_urlopen):
|
|
||||||
"""send_telegram_message raises HTTPError on rate limit (429)."""
|
|
||||||
from urllib.error import HTTPError
|
|
||||||
mock_urlopen.side_effect = HTTPError(
|
|
||||||
url="https://api.telegram.org/bottest_token/sendMessage",
|
|
||||||
code=429,
|
|
||||||
msg="Too Many Requests",
|
|
||||||
hdrs={},
|
|
||||||
fp=None
|
|
||||||
)
|
|
||||||
|
|
||||||
with self.assertRaises(HTTPError) as ctx:
|
|
||||||
send_telegram_message("test_token", "test_chat", "hello")
|
|
||||||
self.assertEqual(ctx.exception.code, 429)
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_network_error_raises_url_error(self, mock_urlopen):
|
|
||||||
"""send_telegram_message raises URLError on network failure."""
|
|
||||||
from urllib.error import URLError
|
|
||||||
mock_urlopen.side_effect = URLError("Connection refused")
|
|
||||||
|
|
||||||
with self.assertRaises(URLError) as ctx:
|
|
||||||
send_telegram_message("test_token", "test_chat", "hello")
|
|
||||||
self.assertIn("Connection refused", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_timeout_raises_url_error(self, mock_urlopen):
|
|
||||||
"""send_telegram_message raises URLError on timeout."""
|
|
||||||
from urllib.error import URLError
|
|
||||||
mock_urlopen.side_effect = URLError("<urlopen error TimeoutError: timed out>")
|
|
||||||
|
|
||||||
with self.assertRaises(URLError):
|
|
||||||
send_telegram_message("test_token", "test_chat", "hello")
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_custom_timeout_used(self, mock_urlopen):
|
|
||||||
"""send_telegram_message respects custom timeout parameter."""
|
|
||||||
mock_resp = MagicMock()
|
|
||||||
mock_resp.read.return_value = b'{"ok": true, "result": {"message_id": 456}}'
|
|
||||||
mock_urlopen.return_value.__enter__.return_value = mock_resp
|
|
||||||
|
|
||||||
send_telegram_message("test_token", "test_chat", "hello", timeout=30)
|
|
||||||
|
|
||||||
call_kwargs = mock_urlopen.call_args[1]
|
|
||||||
self.assertEqual(call_kwargs['timeout'], 30)
|
|
||||||
|
|
||||||
@patch('browse.urlopen')
|
|
||||||
def test_send_html_parsing_mode(self, mock_urlopen):
|
|
||||||
"""send_telegram_message sends with parse_mode=HTML."""
|
|
||||||
mock_resp = MagicMock()
|
|
||||||
mock_resp.read.return_value = b'{"ok": true, "result": {"message_id": 789}}'
|
|
||||||
mock_urlopen.return_value.__enter__.return_value = mock_resp
|
|
||||||
|
|
||||||
send_telegram_message("test_token", "test_chat", "<b>bold</b>")
|
|
||||||
|
|
||||||
call_args = mock_urlopen.call_args
|
|
||||||
req = call_args[0][0]
|
|
||||||
# Verify parse_mode=HTML is in the data
|
|
||||||
self.assertIn(b"parse_mode=HTML", req.data)
|
|
||||||
|
|
||||||
|
|
||||||
class TestHtmlInjection(unittest.TestCase):
|
|
||||||
"""Tests for HTML injection prevention in Telegram messages."""
|
|
||||||
|
|
||||||
@patch.dict('os.environ', {'TELEGRAM_BOT_TOKEN': 'test_token', 'CHAT_ID': 'test_chat'})
|
|
||||||
@patch('browse.send_telegram_message')
|
|
||||||
def test_send_to_telegram_html_injection_in_match_title(self, mock_send_msg):
|
|
||||||
"""
|
|
||||||
titles in match events are NOT escaped before inserting into HTML.
|
|
||||||
This test FAILS if HTML chars are unescaped (vulnerable),
|
|
||||||
and PASSES once escape_html() is implemented.
|
|
||||||
"""
|
|
||||||
mock_send_msg.return_value = 123
|
|
||||||
|
|
||||||
# Simulate a Polymarket event with HTML injection in the title
|
|
||||||
malicious_event = {
|
|
||||||
"title": "<script>alert('XSS')</script> - Team A vs Team B",
|
|
||||||
"slug": "test-event",
|
|
||||||
"startTime": "2027-03-26T12:00:00Z",
|
|
||||||
"markets": [{
|
|
||||||
"sportsMarketType": "moneyline",
|
|
||||||
"outcomes": '["Team A", "Team B"]',
|
|
||||||
"outcomePrices": "[0.55, 0.45]",
|
|
||||||
"bestBid": "0.54",
|
|
||||||
"bestAsk": "0.56",
|
|
||||||
"volume": 50000,
|
|
||||||
"acceptingOrders": True,
|
|
||||||
"closed": False,
|
|
||||||
}],
|
|
||||||
}
|
|
||||||
|
|
||||||
from browse import send_to_telegram
|
|
||||||
send_to_telegram([malicious_event], [], "Counter Strike")
|
|
||||||
|
|
||||||
# Check what was passed to send_telegram_message
|
|
||||||
self.assertEqual(mock_send_msg.called, True)
|
|
||||||
sent_text = mock_send_msg.call_args[0][2] # text arg (3rd positional)
|
|
||||||
|
|
||||||
# AFTER FIX: <script> should be escaped as <script>
|
|
||||||
# BEFORE FIX: raw <script> appears in text (vulnerable — test would fail here)
|
|
||||||
self.assertIn("<script>", sent_text,
|
|
||||||
"HTML injection still present — title may NOT be escaped")
|
|
||||||
self.assertIn("</script>", sent_text)
|
|
||||||
|
|
||||||
@patch.dict('os.environ', {'TELEGRAM_BOT_TOKEN': 'test_token', 'CHAT_ID': 'test_chat'})
|
|
||||||
@patch('browse.send_telegram_message')
|
|
||||||
def test_send_to_telegram_ampersand_in_title(self, mock_send_msg):
|
|
||||||
"""
|
|
||||||
Ampersands in titles should be escaped as & when using HTML parse_mode.
|
|
||||||
BEFORE fix: "&" appears raw in the HTML (vulnerable).
|
|
||||||
AFTER fix: "&" appears as "&".
|
|
||||||
"""
|
|
||||||
mock_send_msg.return_value = 123
|
|
||||||
|
|
||||||
event_with_ampersand = {
|
|
||||||
"title": "Team A & Team B vs Team C",
|
|
||||||
"slug": "amp-test",
|
|
||||||
"startTime": "2027-03-26T12:00:00Z",
|
|
||||||
"markets": [{
|
|
||||||
"sportsMarketType": "moneyline",
|
|
||||||
"outcomes": '["Team A & Team B", "Team C"]',
|
|
||||||
"outcomePrices": "[0.50, 0.50]",
|
|
||||||
"bestBid": "0.49",
|
|
||||||
"bestAsk": "0.51",
|
|
||||||
"volume": 10000,
|
|
||||||
"acceptingOrders": True,
|
|
||||||
"closed": False,
|
|
||||||
}],
|
|
||||||
}
|
|
||||||
|
|
||||||
from browse import send_to_telegram
|
|
||||||
send_to_telegram([event_with_ampersand], [], "Dota 2")
|
|
||||||
|
|
||||||
sent_text = mock_send_msg.call_args[0][2]
|
|
||||||
|
|
||||||
# AFTER FIX: & should be escaped as &
|
|
||||||
# BEFORE FIX: raw & appears (vulnerable — test would fail here)
|
|
||||||
self.assertIn("&", sent_text,
|
|
||||||
"Ampersand not escaped — title may NOT be escaped")
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
Reference in New Issue
Block a user