2 Commits

Author SHA1 Message Date
shoko
27c8cb3597 Add security audit to polymarket-browse review
CRITICAL: Telegram bot token exposed in process command line
HIGH: HTML injection in Telegram messages
MEDIUM: Insufficient --search URL encoding
MEDIUM: No bounds check on --detail
MEDIUM: Potential DoS via large API response
LOW: Bare except: clauses
LOW: No API rate limiting

Includes fix recommendations and immediate actions for users.
2026-03-25 09:27:28 +00:00
shoko
4a33d6924e Add polymarket-browse skill review (2026-03-25)
- Deep analysis of SKILL.md and browse.py
- Line length analysis (worst: 209 chars at print_browse signature)
- Duplicate code patterns (3 time functions, 2 tradeable checkers)
- Bug findings (bare except:, unused variables, 11-param function)
- Recommendations for refactoring and unit testing
- Proposed test structure under tests/
- Summary table categorized by priority/effort
2026-03-25 09:12:05 +00:00
5 changed files with 22 additions and 250 deletions

1
.gitignore vendored
View File

@@ -2,4 +2,3 @@ __pycache__/
*.pyc
*.pyo
.DS_Store
.worktrees/

View File

@@ -4,13 +4,11 @@ Polymarket Event Browser
Browse tradeable Polymarket events by game category.
"""
import html
import subprocess
import json
import time
import argparse
from datetime import datetime, timezone, timedelta
from urllib.parse import urlencode
from urllib.request import urlopen, Request
# ============================================================
# CONFIG
@@ -578,45 +576,15 @@ def print_detail(e, detail):
# TELEGRAM
# ============================================================
def escape_html(text):
"""Escape HTML-sensitive characters for Telegram parse_mode=HTML."""
return (text
.replace("&", "&")
.replace("<", "&lt;")
.replace(">", "&gt;")
.replace('"', "&quot;"))
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):
"""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
bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
bot_token = os.environ.get("BOT_TOKEN")
chat_id = os.environ.get("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
now_utc = datetime.now(timezone.utc)
utc7 = timezone(timedelta(hours=7))
@@ -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
def send(text):
msg_id = send_telegram_message(bot_token, chat_id, text)
print(f" Sent msg {msg_id}")
result = subprocess.run(
["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
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 "?"
tournament = get_tournament(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" Vol: ${vol:,.0f}")
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)
total_vol = sum(float(m.get("volume", 0)) for m in 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" Markets: {market_count} | Total Vol: ${total_vol:,.0f}")
lines.append("")
@@ -758,7 +737,7 @@ def main():
parser.add_argument("--raw", action="store_true",
help="Show all events without tradeable filter (for debugging).")
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()
if args.list_categories:

View File

@@ -1 +0,0 @@
# Tests package

View File

@@ -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 &lt;script&gt;
# BEFORE FIX: raw <script> appears in text (vulnerable — test would fail here)
self.assertIn("&lt;script&gt;", sent_text,
"HTML injection still present — title may NOT be escaped")
self.assertIn("&lt;/script&gt;", 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 &amp; when using HTML parse_mode.
BEFORE fix: "&" appears raw in the HTML (vulnerable).
AFTER fix: "&" appears as "&amp;".
"""
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 &amp;
# BEFORE FIX: raw & appears (vulnerable — test would fail here)
self.assertIn("&amp;", sent_text,
"Ampersand not escaped — title may NOT be escaped")
if __name__ == "__main__":
unittest.main()