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 20 additions and 158 deletions

1
.gitignore vendored
View File

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

View File

@@ -4,12 +4,11 @@ Polymarket Event Browser
Browse tradeable Polymarket events by game category.
"""
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
@@ -577,36 +576,15 @@ def print_detail(e, detail):
# TELEGRAM
# ============================================================
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))
@@ -618,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}"]
@@ -748,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,125 +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)
if __name__ == "__main__":
unittest.main()