9 Commits

Author SHA1 Message Date
shoko
a7837cec0f Merge #15: Unify duplicate time functions 2026-03-25 14:34:05 +00:00
shoko
8cde441996 Fix #15: Unify duplicate time functions into _get_time_data()
Replace three duplicated time parsing functions with a single
_get_time_data(e, tz) helper returning {time_status, time_urgency, abs_time}.

Deleted functions:
- get_match_time_status(e)  — urgency + status string
- get_match_time_str(e)    — status string only
- get_start_time_wib(e)    — (abs_time, rel_str) tuple

New unified helper:
- _get_time_data(e, tz=None) returns {time_status, time_urgency, abs_time}
- tz defaults to WIB (UTC+7, Indonesia)
- canonical rel_str format: 'LIVE', 'In 6h', '12h ago', etc.
- time_urgency: 0-3 (higher=livelier)

All call sites updated to use _get_time_data():
- format_event(), format_detail_event()
- print_browse(), print_detail()
- send_to_telegram()

Also: removed dead code in print_detail() that called get_match_time_str()
but never used the result.

Tests: 9 new tests for _get_time_data() covering TBD, future, live,
and past event scenarios. 19 tests total, all passing.

Fixes: #15
2026-03-25 13:59:54 +00:00
b2180a4a34 Merge pull request 'Fix #5: HTML injection in Telegram messages' (#20) from fix/issue-5-html-injection-telegram into master 2026-03-25 13:13:52 +01:00
shoko
d0534aedbf Fix #5: HTML injection in Telegram messages
Add escape_html() function to prevent HTML injection in Telegram
parse_mode=HTML messages. Apply escaping to event titles inserted
into <a> tags in send_to_telegram().

- Add escape_html() using stdlib html.escape()
- Escape match event titles (line 648) and non-match titles (line 676)
- Add TestHtmlInjection with 2 tests proving fix:
  - <script> tags escaped as &lt;script&gt;
  - & ampersands escaped as &amp;
- Fixes HIGH severity: titles from Polymarket API were inserted
  without escaping, allowing malformed HTML in Telegram messages
2026-03-25 11:42:42 +00:00
2703b942c1 Merge pull request 'Fix #4: Extract send() to module-level send_telegram_message() for testability' (#19) from fix/issue-4-telegram-token-refactor into master 2026-03-25 12:17:00 +01:00
shoko
f9c4bac7b8 Refactor send() to module-level send_telegram_message() for testability
Extract the nested send() function into a module-level
send_telegram_message(bot_token, chat_id, text, timeout=10)
function. This enables unit testing without hitting the real
Telegram API.

Changes:
- Add send_telegram_message() at module level in TELEGRAM section
- Replace nested send() with thin wrapper that calls
  send_telegram_message()
- Update argparse --telegram help text to use TELEGRAM_BOT_TOKEN
- Add tests/test_browse.py with 8 unit tests covering:
  - Success case (returns message_id)
  - API error (RuntimeError)
  - Invalid token (HTTPError 404)
  - Rate limit (HTTPError 429)
  - Network error (URLError)
  - Timeout (URLError)
  - Custom timeout parameter
  - HTML parse_mode in request

Ref: #4
2026-03-25 11:07:10 +00:00
shoko
c49600cd4d Fix CRITICAL: Telegram bot token exposed in process command line
Replace curl subprocess with urllib.request to prevent token leakage via
ps aux / /proc/*/cmdline. Token now stays in process memory only.

Changes:
- Remove subprocess import, add urllib.parse.urlencode and urllib.request
- Replace curl subprocess call with urlopen(Request(...))
- Change env var BOT_TOKEN -> TELEGRAM_BOT_TOKEN (clearer naming)
- Raise RuntimeError on missing env vars, API errors, or network errors
- Add 10s timeout to urlopen

Fixes #4
2026-03-25 10:46:10 +00:00
shoko
3a988943b9 docs: rename review folder to match skill structure
docs/polymarket-browse/ mirrors skills/polymarket-browse/
Future reviews for this skill can use date-based filenames in the same folder.
2026-03-25 10:02:43 +00:00
shoko
da367c594b docs: add polymarket-browse review (2026-03-25)
Security audit + code quality review of polymarket-browse skill.
Contains 8 security issues, 6 code quality issues, 2 docs issues.
Issues tracked in repo.
2026-03-25 10:00:12 +00:00
5 changed files with 1231 additions and 158 deletions

1
.gitignore vendored
View File

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

View File

@@ -0,0 +1,778 @@
# Polymarket-Browse Skill Review
**Date:** 2026-03-25
**Reviewer:** Hermes Agent (Shoko)
**Version Reviewed:** Current HEAD
---
## 1. Current State of SKILL.md
### 1.1 Overview
The SKILL.md is well-structured with clear sections:
- Installation instructions (Hermes Agent + OpenClaw)
- Usage with argument reference
- Output format examples
- Game categories table
- Filters explanation
- Pagination and rate limiting notes
- Odds format documentation
### 1.2 Strengths
- Clear argument documentation with defaults
- Good output format examples showing both match and non-match markets
- Filters section is detailed and explains tradeable vs non-tradeable logic
- Game categories table is easy to reference
- Rate limiting and backoff strategy documented
### 1.3 Issues/Gaps in SKILL.md
| Issue | Severity | Notes |
|-------|----------|-------|
| No troubleshooting section | Low | API errors, partial fetches, common issues not documented |
| No examples for --search | Low | Only mentioned in passing, no concrete example |
| No mention of required dependencies | Low | Assumes curl is available (it is on Linux) |
| No changelog | Low | Hard to track what changed between versions |
| Telegram section minimal | Low | Doesn't explain HTML parse_mode limitations |
| No credits/author info | Low | Who built this? |
### 1.4 Recommendations for SKILL.md
1. **Add troubleshooting section:**
- Partial fetch warnings (API errors/timeout)
- What to do if no markets appear
- Explaining why some matches disappear after they start
2. **Add concrete usage examples:**
```bash
# Example: Find FlyQuest Counter-Strike matches
polymarket-browse --category "Counter Strike" --search "FlyQuest"
# Example: Get 10 matches, no tournament futures
polymarket-browse --category "Valorant" --matches 10 --non-matches-only
```
3. **Add HTML escape notes for Telegram:**
- `<` `>` `&` `>` `<` need to be escaped in Telegram messages
---
## 2. Current State of browse.py
### 2.1 Code Organization
The script is organized into logical sections with clear headers:
```
CONFIG
FETCH
FILTERS
FORMATTING
BROWSE
FORMAT
DISPLAY
TELEGRAM
MAIN
```
**Issues:**
- Lines are excessively long (erowse ~750 lines, some functions are very dense)
- `print_browse()` function is ~120 lines — too long to review mentally
- `send_to_telegram()` function is ~100 lines — also too long
- `format_detail_event()` has deeply nested list comprehensions
- No type hints anywhere
- No docstrings on main functions (only on helper functions)
### 2.2 Line Length Issues (CRITICAL)
The user specifically asked about this. Here are the longest lines:
| Line | Length | Issue |
|------|--------|-------|
| ~line 100 | ~180 chars | `fetch_page()` URL construction |
| ~line 160 | ~160 chars | `fetch_all_pages()` loop |
| ~line 210 | ~200 chars | `is_tradeable_event()` ML market checks |
| ~line 240 | ~180 chars | `is_tradeable_event()` datetime parsing |
| ~line 300 | ~180 chars | `get_match_time_status()` datetime math |
| ~line 380 | ~200 chars | `format_detail_event()` list comprehension |
| ~line 470 | ~220 chars | `print_browse()` event formatting |
| ~line 540 | ~180 chars | `send_to_telegram()` message building |
**Root cause:** The code was written for functionality, not readability. String concatenation and nested conditionals make lines very long.
### 2.3 Proposed Solutions for Line Length
**Option A: Refactor to shorter lines (Recommended)**
- Break long URL constructions into multiple lines
- Extract nested conditionals into helper variables
- Use intermediate variables for complex expressions
- Target: max 120 characters per line
**Option B: Add unit tests**
- Write unit tests that verify behavior without needing to read every line
- Tests serve as executable documentation
- Anyone can run `pytest` to verify correctness
- See Section 3 for details
**Option C: Both (Recommended)**
- Refactor for readability
- Add unit tests for critical paths
- This is the best approach
### 2.4 Function-by-Function Analysis
#### `fetch_page()` (~35 lines)
**What it does:** Fetches one page from Polymarket API with retry logic
**Issues:**
- URL construction is on one long line
- Exponential backoff is clear but verbose
- Could use `requests` library instead of curl subprocess
**Suggestions:**
- Break URL construction into multiple `params = {...}` style
- Consider using `httpx` or `requests` instead of curl subprocess
#### `fetch_all_pages()` (~25 lines)
**What it does:** Paginates through all results
**Issues:**
- `time.sleep(0.2)` is hardcoded — should be configurable
- No progress indicator for large fetches
**Suggestions:**
- Add progress callback option
- Make inter-page delay configurable
#### `is_tradeable_event()` (~70 lines)
**What it does:** Complex filter for tradeable match markets
**Issues:**
- This is the longest function at ~70 lines
- Multiple filter conditions stacked vertically (good) but with long lines (bad)
- Bare `except:` clauses that catch everything
**Suggestions:**
- Extract `is_bo2_tie()` check (already done — good)
- Extract datetime comparisons into helper functions
- Add early returns to reduce nesting
- Change bare `except:` to specific exceptions
#### `is_tradeable_market()` (~20 lines)
**What it does:** Filter for individual markets
**Issues:**
- Very similar to `is_tradeable_event()` — code duplication
- Could reuse logic from the event version
**Suggestions:**
- Consider unifying with `is_tradeable_event()`
#### `get_match_time_status()` / `get_match_time_str()` (~50 lines combined)
**What it does:** Time formatting for display
**Issues:**
- Duplicate logic — both functions do similar things
- WIB (UTC+7) is hardcoded — user is Indonesian, but this should be configurable
**Suggestions:**
- Consolidate into one function that returns both values
- Make timezone configurable
#### `print_browse()` (~120 lines)
**What it does:** Main display function for CLI output
**Issues:**
- ~120 lines is too long to review mentally
- Mixes display logic with data formatting
- Has its own datetime import (Python import inside function — anti-pattern)
**Suggestions:**
- Break into smaller functions:
- `format_match_line()`
- `format_non_match_line()`
- `print_match_section()`
- `print_non_match_section()`
#### `send_to_telegram()` (~100 lines)
**What it does:** Telegram integration
**Issues:**
- ~100 lines too long
- Complex chunking logic for Telegram 4096 char limit
- HTML escaping not handled
**Suggestions:**
- Extract chunking logic into separate function
- Add HTML escaping helper
- Consider using `python-telegram-bot` library instead of curl
#### `format_detail_event()` (~30 lines)
**What it does:** Formats event with all markets for detail view
**Issues:**
- List comprehension is deeply nested and hard to read
- ~15-line dict construction
**Suggestions:**
- Break the dict construction into multiple lines
- Extract market formatting into helper
### 2.5 Error Handling
| Issue | Severity | Notes |
|-------|----------|-------|
| Bare `except:` clauses | Medium | Catches KeyboardInterrupt, SystemExit |
| No logging | Low | Uses print statements |
| No structured errors | Low | Could benefit from custom exceptions |
### 2.6 Missing Features/Bugs
| Issue | Severity | Notes |
|-------|----------|-------|
| No test suite | High | Cannot verify correctness automatically |
| WIB hardcoded | Medium | Timezone should be configurable |
| No cache option | Low | Could cache results for repeated queries |
| `--detail` uses 1-indexed but docs unclear | Low | Works, but could be confusing |
| BO2 tie detection uses title match | Medium | Relies on "BO2" in title — fragile |
| `is_bo2_tie()` checks child_moneyline closed | Medium | API may not always set this flag |
---
## 3. Recommended Improvements
### 3.1 Code Refactoring (Priority: HIGH)
**Goal:** Make browse.py reviewable by humans
**Specific changes:**
1. **Break `print_browse()` into helper functions:**
```python
def format_match_line(i, e, ml, outcomes, prices, vol, title, url, ...):
"""Format a single match event line."""
...
def print_match_section(match_events, ...):
"""Print the MATCH MARKETS section."""
...
```
2. **Break `send_to_telegram()` into helper functions:**
```python
def escape_html(text):
"""Escape HTML special characters."""
...
def chunk_telegram_message(lines, max_len=4096):
"""Split long messages into chunks."""
...
```
3. **Break long lines:**
- URL construction: use `params = {...}` dict style
- Long conditionals: extract to named variables
- Long f-strings: break across multiple lines
4. **Add type hints:**
```python
def fetch_page(q: str, page: int = 1, ...) -> Optional[dict]:
```
5. **Consolidate duplicate time functions:**
- `get_match_time_status()` and `get_match_time_str()` share logic
- Create one function returning both
### 3.2 Unit Tests (Priority: HIGH)
**Goal:** Enable human review via test execution, not line-by-line reading
**Proposed test structure:**
```
tests/
__init__.py
test_filters.py # is_match_market, is_tradeable_event, is_tradeable_market
test_formatters.py # format_odds, prob_to_cents, get_match_time_*
test_browse.py # Integration tests with mocked API
test_cli.py # Argument parsing tests
```
**Test examples:**
```python
# test_formatters.py
def test_prob_to_cents():
assert prob_to_cents(0.30) == 30
assert prob_to_cents(0.95) == 95
assert prob_to_cents(0.001) == 0
def test_format_odds():
assert format_odds(0.30) == "30c"
assert format_odds(0.95) == "95c"
# test_filters.py
def test_is_match_market_with_series():
e = {"seriesSlug": "csg", "gameId": "123", "title": "Team A vs Team B"}
assert is_match_market(e) == True
def test_is_match_market_vs_syntax():
e = {"title": "Team A vs Team B"}
assert is_match_market(e) == True
def test_is_match_market_non_match():
e = {"title": "Tournament Winner"}
assert is_match_market(e) == False
# test_filters.py - is_tradeable_event
def test_bo2_tie_filter():
"""BO2 matches ending 1-1 should be filtered out."""
e = create_bo2_event(ended_tie=True)
assert is_tradeable_event(e) == False
def test_converged_market_filter():
"""Market with bestBid >= 0.99 should be filtered."""
e = create_event_with_ml(bestBid=0.99, bestAsk=0.99)
assert is_tradeable_event(e) == False
```
**Mock API responses needed:**
- Store sample API responses in `tests/fixtures/` as JSON
- Use `responses` or `requests-mock` to mock HTTP calls
### 3.3 Documentation Improvements (Priority: MEDIUM)
1. Add troubleshooting section to SKILL.md
2. Add concrete usage examples
3. Add HTML escape notes for Telegram
4. Add changelog
5. Document the 1-indexed `--detail` argument more clearly
### 3.4 Configuration Options (Priority: LOW)
1. Make timezone (WIB) configurable via `--timezone` argument or env var
2. Make inter-page delay configurable
3. Add `--json` output option for programmatic use
---
## 4. Summary Table
| Category | Item | Priority | Effort |
|----------|------|----------|--------|
| **Code** | Refactor print_browse() into smaller functions | HIGH | Medium |
| **Code** | Refactor send_to_telegram() into smaller functions | HIGH | Medium |
| **Code** | Break long lines to max 120 chars | HIGH | Low |
| **Tests** | Add unit tests for filters | HIGH | Medium |
| **Tests** | Add unit tests for formatters | HIGH | Low |
| **Tests** | Add integration tests with mocked API | MEDIUM | Medium |
| **Docs** | Add troubleshooting section to SKILL.md | MEDIUM | Low |
| **Docs** | Add usage examples to SKILL.md | MEDIUM | Low |
| **Code** | Consolidate duplicate time functions | LOW | Low |
| **Code** | Add type hints | LOW | Medium |
| **Config** | Make timezone configurable | LOW | Low |
---
## 5. Next Steps
1. **Immediate:** Create unit test structure under `tests/`
2. **Short-term:** Refactor `print_browse()` and `send_to_telegram()` into smaller functions
3. **Short-term:** Break long lines to max 120 characters
4. **Medium-term:** Add comprehensive unit tests
5. **Medium-term:** Update SKILL.md with troubleshooting and examples
---
---
## 6. Security Audit
### 6.1 Severity Classification
| Severity | Meaning |
|----------|---------|
| CRITICAL | Immediate action required. Users should stop using the skill until fixed. |
| HIGH | Serious vulnerability. Fix as soon as possible. |
| MEDIUM | Moderate issue. Fix in normal development cycle. |
| LOW | Minor issue. Fix when convenient. |
---
### 6.2 CRITICAL: Telegram Bot Token Exposed in Process Listings
**Location:** Lines 598-606 (`send_to_telegram()`)
**Description:**
The Telegram bot token is passed as a subprocess argument to `curl`, making it visible in the process command line. On any Unix system, any user can view all running processes' command lines via `ps aux` or `/proc/*/cmdline`.
**Proof of Concept:**
```bash
# While the script is running (or just after it finishes):
ps aux | grep curl
# Output reveals token:
# curl -s https://api.telegram.org/bot123456789:ABCdefGHI/sendMessage -d chat_id=... -d text=...
```
**Impact:**
- Any user on a shared system can steal the bot token
- Attacker can send arbitrary messages from the bot
- Attacker can use the bot for spam, phishing, or social engineering
- If the bot is in group chats, attacker can extract member information
**Fix:**
Use curl with `--oauth2-bearer` or environment variable approach. The bot token should NEVER appear in the command line. Recommended fix:
```python
import os
def send_to_telegram(...):
bot_token = os.environ.get("BOT_TOKEN")
chat_id = os.environ.get("CHAT_ID")
if not bot_token or not chat_id:
return
env = os.environ.copy()
# Use a temp file or proper curl auth method
# Actually Telegram bot tokens cannot be passed via header
# Instead: use Python's requests library which handles this securely
# OR: write token to a file with 0600 perms and use curl's --config option
```
**Proper fix using Python `requests` library:**
```python
import requests
def send_to_telegram(...):
bot_token = os.environ.get("BOT_TOKEN")
chat_id = os.environ.get("CHAT_ID")
if not bot_token or not chat_id:
return
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
payload = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"}
# Token stays in memory, never in command line
resp = requests.post(url, data=payload, timeout=10)
```
**Interim mitigation:** If you must use curl, write the token to a temporary file with mode 0600 and use curl's `--config` flag, then delete the file immediately after.
---
### 6.3 HIGH: HTML Injection in Telegram Messages (XSS-adjacent)
**Location:** Lines 614-661 (`send_to_telegram()`)
**Description:**
Event titles and other data from the Polymarket API are inserted directly into Telegram messages with `parse_mode=HTML`. No HTML escaping is performed. Characters like `<`, `>`, `&` are not escaped.
**Attack scenario:**
1. Polymarket (or an attacker who compromises Polymarket data) includes a malicious title like:
- `<script>alert('XSS')</script>`
- `<img src=x onerror=alert(1)>`
- `Team A &amp; Team B` (ampersand not escaped renders as `&amp;amp;` or breaks parsing)
2. The bot sends this to Telegram
3. Telegram renders the HTML, potentially executing JavaScript in the context of the Telegram web client
**Note:** Telegram's HTML parser is restrictive (`<b>`, `<i>`, `<a>`, etc. only), so classic XSS is limited. However:
- Malformed HTML can crash the Telegram client
- Attribute-based injections in `<a>` tags could be possible
- The `<` and `>` characters themselves are illegal in Telegram HTML and will cause parse errors
**Current vulnerable code (simplified):**
```python
title = e.get("title", "?")
lines.append(f"<b>{i}.</b> <a href=\"{url}\">{title_clean}</a>")
# title_clean is title.split(" - ")[0].strip() -- no HTML escaping
```
**Fix:**
```python
import html
def escape_html(text):
"""Escape HTML special characters for Telegram."""
return (text
.replace("&", "&amp;")
.replace("<", "&lt;")
.replace(">", "&gt;")
.replace('"', "&quot;"))
title = e.get("title", "?")
title_escaped = escape_html(title_clean)
lines.append(f"<b>{i}.</b> <a href=\"{url}\">{title_escaped}</a>")
```
---
### 6.4 MEDIUM: Insufficient Input Sanitization on `--search` Parameter
**Location:** Line 39 (`fetch_page()`)
**Description:**
The `--search` argument is only sanitized with `.replace(' ', '%20')`. This only handles spaces. Other URL-sensitive characters (`#`, `?`, `&`, `%`, `+`, etc.) are not properly URL-encoded.
**Attack scenario:**
A user passes `--search "Team%20A"` expecting to search for "Team%20A" literally, but `%20` gets decoded to a space by the URL parser, searching for "Team A" instead.
More concerning: if the search term contains `#`, it could truncate the query:
```bash
--search "Team#A" -> URL becomes ?q=Team#A&limit=50...
# Everything after # is treated as URL fragment, not part of the query
```
**Current code:**
```python
url = (f"{base}?q={q.replace(' ', '%20')}&limit={PAGE_SIZE}&page={page}"
f"&search_profiles=false&search_tags=false"
f"&keep_closed_markets=0&events_status=active&cache=false")
```
**Fix:**
Use `urllib.parse.quote()` for proper URL encoding:
```python
from urllib.parse import quote
def fetch_page(q, page=1, ...):
base = "https://gamma-api.polymarket.com/public-search"
encoded_q = quote(q, safe='')
url = (f"{base}?q={encoded_q}&limit={PAGE_SIZE}&page={page}"
...)
```
---
### 6.5 MEDIUM: No Bounds Checking on `--detail` Argument
**Location:** Lines 778-785 (`main()`)
**Description:**
The `--detail N` argument is used to index into `result["match_events"]` without proper bounds checking. While there is a fallback (`idx = 0` if out of range), the logic silently defaults to index 0, which may not be what the user intended.
```python
idx = args.detail - 1 # User provides 1-indexed
if idx < 0 or idx >= len(result["match_events"]):
idx = 0 # Silently defaults to first event
detail_event = result["match_events"][idx]
```
**Impact:**
- Low security impact, but confusing UX
- User expects event #100 but gets event #1 silently
**Fix:** Warn user if index is out of range instead of silently defaulting.
---
### 6.6 MEDIUM: Potential Denial of Service via Large API Response
**Location:** Lines 53, 79-81, 357-358, 393-403
**Description:**
The code loads the entire API response into memory via `json.loads()`. If a malicious or compromised Polymarket API returned an extremely large JSON payload (gigabytes), the script could exhaust system memory.
**Additional issue:** `fetch_all_pages()` loops up to 100 pages, each with up to 50 events. While bounded, there's no size limit on individual events or their nested `markets` arrays.
**Fix:**
```python
# Add response size limits
MAX_RESPONSE_SIZE = 10 * 1024 * 1024 # 10MB
if len(r.stdout) > MAX_RESPONSE_SIZE:
raise ValueError(f"API response too large: {len(r.stdout)} bytes")
```
---
### 6.7 LOW: Bare `except:` Clauses Swallow Security-Relevant Errors
**Location:** Lines 169, 183, 269, 308, 456
**Description:**
Bare `except: pass` catches ALL exceptions including `KeyboardInterrupt`, `SystemExit`, `MemoryError`, and `OSError`. This silently hides errors that might indicate security problems (e.g., timeouts could suggest a DoS attack in progress).
**Current code:**
```python
try:
end_dt = datetime.fromisoformat(end_str.replace('Z', '+00:00'))
...
except:
pass # Silently ignores ALL errors
```
**Fix:**
```python
try:
end_dt = datetime.fromisoformat(end_str.replace('Z', '+00:00'))
...
except (ValueError, TypeError):
pass # Only catch expected exceptions
```
---
### 6.8 LOW: No Rate Limiting on API Calls (DoS vector)
**Location:** Lines 74-76 (`fetch_all_pages()`)
**Description:**
An attacker who can pass arguments to this script (e.g., via a web wrapper) could repeatedly call the Polymarket API in a loop, potentially:
1. Getting the user's IP rate-limited or banned by Polymarket
2. Consuming the user's bandwidth
3. Causing resource exhaustion on the host
**Fix:** Add a cooldown between runs if called repeatedly, or use a token bucket rate limiter.
---
### 6.9 Security Summary Table
| # | Issue | Severity | Exploitable Remotely | Fix Complexity |
|---|-------|----------|----------------------|----------------|
| 1 | Telegram bot token in process cmdline | CRITICAL | No (local access needed) | Easy |
| 2 | HTML injection in Telegram messages | HIGH | No (needs Polymarket compromise) | Easy |
| 3 | Insufficient `--search` sanitization | MEDIUM | Yes (any user input) | Easy |
| 4 | No bounds check on `--detail` | MEDIUM | Yes (any user input) | Trivial |
| 5 | Large API response can exhaust memory | MEDIUM | Yes (API or MITM) | Easy |
| 6 | Bare `except:` swallows errors | LOW | No | Trivial |
| 7 | No API rate limiting | LOW | Yes (with script access) | Medium |
---
### 6.10 Recommended Immediate Actions
**If you are currently using this skill with a Telegram bot:**
1. **ROTATE YOUR BOT TOKEN IMMEDIATELY** — Go to @BotFather and use `/revoke` to invalidate the current token. Generate a new one.
2. **Do not use the `--telegram` flag** on shared/multi-user systems until Issue #1 is fixed.
3. **Do not use `--search` with untrusted input** until Issue #3 is fixed.
4. Monitor your Telegram bot's `getUpdates` for unauthorized access.
**Safe usage until fixes are deployed:**
```bash
# Without Telegram (safe):
polymarket-browse --category "Counter Strike" --limit 5
# With Telegram (UNSAFE on shared systems until fix is deployed):
# NOT RECOMMENDED until security issues are addressed
```
---
### 6.11 Fix Priority Order
1. **FIRST (CRITICAL):** Fix Telegram bot token exposure — use Python `requests` library instead of curl subprocess, OR use curl with `--config` file approach
2. **SECOND (HIGH):** Add HTML escaping for Telegram messages
3. **THIRD (MEDIUM):** Fix `--search` URL encoding with `urllib.parse.quote()`
4. **FOURTH (MEDIUM):** Add `--detail` bounds checking
5. **FIFTH (MEDIUM):** Add response size limits
6. **SIXTH (LOW):** Replace bare `except:` with specific exceptions
---
### Appendix A: Longest Lines in browse.py (for targeted refactoring)
| Line | Chars | Location | Content Summary |
|------|-------|----------|-----------------|
| 474 | 209 | `print_browse()` | Function signature |
| 564 | 152 | `print_detail()` | ML odds formatting |
| 571 | 136 | `print_detail()` | Market outcome formatting |
| 760 | 128 | `send_to_telegram()` | Telegram send call |
| 561 | 126 | `print_detail()` | Spread formatting |
| 736 | 122 | `send_to_telegram()` | Telegram API URL |
| 485 | 121 | `print_browse()` | Fetch stats line |
| 467 | 119 | `print_browse()` | Print category header |
| 728 | 112 | `send_to_telegram()` | Telegram send call |
| 569 | 110 | `print_detail()` | Market spread formatting |
**Key finding:** The `print_browse()` function signature itself (line 474) at 209 chars is the longest. This should be broken up or the function should accept a config dict instead of 11 parameters.
---
## Appendix B: Duplicate Code Patterns
### B.1 Time formatting duplicated across 3 functions
| Function | Lines | Purpose |
|----------|-------|---------|
| `get_match_time_status()` | ~40 | Returns (status_str, urgency) tuple |
| `get_match_time_str()` | ~35 | Returns just status string |
| `get_start_time_wib()` | ~50 | Returns (abs_str, rel_str) tuple |
All three parse the same ISO datetime string and compute the same relative time logic. Should be consolidated into one function returning all needed values.
### B.2 `is_tradeable_event()` vs `is_tradeable_market()`
Both check convergence (bestBid >= 0.99, bestAsk <= 0.01) and acceptingOrders/closed status. The market-level one is simpler but they share the same convergence check logic.
---
## Appendix C: Potential Bugs
### C.1 Bare `except:` clauses
Found at lines 169, 183, and similar locations:
```python
except:
pass
```
**Risk:** Catches KeyboardInterrupt, SystemExit, and json.JSONDecodeError. Should be:
```python
except (ValueError, TypeError):
pass
```
### C.2 Line 474: `print_browse()` signature is 209 characters
```python
def print_browse(match_events, non_match_events, category, total_raw, total_fetched, total_match, total_non_match, raw_mode=False, partial=False, non_matches_max=5, matches_only=False, non_matches_only=False):
```
**Issue:** 11 parameters is too many. Consider using a result dict or a config object.
**Fix options:**
1. Accept a `BrowseResult` namedtuple/dataclass
2. Split into `print_browse_header()` and `print_browse_sections()`
3. Use `**kwargs`
### C.3 Line 128 in `send_to_telegram()`: `bot_token=os.environ.get("BOT_` (truncated)
```python
bot_token=os.environ.get("BOT_TOKEN")
chat_id = os.environ.get("CHAT_ID")
```
This looks like a line that was cut off in the output but the actual code is fine. However, it highlights that the line at 582 is long.
### C.4 `chunk_len` variable unused
At line 681 in `send_to_telegram()`:
```python
chunk = []
chunk_len = 0 # NEVER USED
chunk_num = 1 # NEVER USED
```
---
## Appendix D: Missing Test Coverage
Functions that need tests but have none:
```
[x] fetch_page - needs mock curl response
[x] fetch_all_pages - needs mock paginated responses
[x] is_match_market - easy to test with dict inputs
[x] is_tradeable_event - complex, needs many test cases
[x] is_tradeable_market - similar to above
[x] is_bo2_tie - edge cases for BO2 detection
[x] get_ml_market - easy to test
[x] get_ml_volume - easy to test
[x] prob_to_cents - pure function, easy to test
[x] format_odds - pure function, easy to test
[x] format_spread - pure function, easy to test
[x] get_match_time_* - needs timezone mocking
[x] get_tournament - easy to test
[x] get_event_url - easy to test
[x] filter_events - easy to test
[x] sort_events - easy to test
```
---
*Report generated by Hermes Agent on 2026-03-25*

View File

@@ -4,11 +4,13 @@ Polymarket Event Browser
Browse tradeable Polymarket events by game category. Browse tradeable Polymarket events by game category.
""" """
import subprocess import html
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
@@ -17,6 +19,7 @@ from datetime import datetime, timezone, timedelta
PAGE_SIZE = 50 PAGE_SIZE = 50
MAX_RETRIES = 5 MAX_RETRIES = 5
INITIAL_RETRY_DELAY = 2 # exponential backoff starts at 2s INITIAL_RETRY_DELAY = 2 # exponential backoff starts at 2s
WIB = timezone(timedelta(hours=7)) # UTC+7 for Indonesian users
GAME_CATEGORIES = { GAME_CATEGORIES = {
"All Esports": "Esports", "All Esports": "Esports",
@@ -219,94 +222,79 @@ def format_spread(bid, ask):
spread = ask - bid spread = ask - bid
return f"{prob_to_cents(spread)}c" return f"{prob_to_cents(spread)}c"
def get_match_time_status(e):
def _get_time_data(e, tz=None):
""" """
Return a human-readable match time status. Unified time data extraction for event timestamps.
Returns (status_str, urgency) where urgency is 0-3 (higher = more urgent/live).
Uses startTime for actual match start time. Uses startTime (preferred) or startDate as the event start time.
Displays times in WIB (UTC+7 for Indonesian users). Datetime parsing and all relative calculations are UTC-based.
The tz parameter only affects the abs_time formatting.
Args:
e: Event dict with 'startTime' or 'startDate' key.
tz: datetime.timezone for abs_time formatting.
Defaults to WIB (UTC+7).
Returns:
{
"time_status": str, # e.g. "LIVE", "In 6h", "12h ago"
"time_urgency": int, # 0-3 (higher = more urgent/live)
"abs_time": str, # e.g. "Mar 25, 19:00 WIB" or "TBD"
}
""" """
# Use startTime for actual match start, not startDate (which is market creation time) tz = tz or WIB
start_str = e.get("startTime") or e.get("startDate", "") start_str = e.get("startTime") or e.get("startDate", "")
if not start_str: if not start_str:
return "TBD", 0 return {"time_status": "TBD", "time_urgency": 0, "abs_time": "TBD"}
try: try:
start_dt = datetime.fromisoformat(start_str.replace('Z', '+00:00')) start_dt = datetime.fromisoformat(start_str.replace('Z', '+00:00'))
now_utc = datetime.now(timezone.utc) now_utc = datetime.now(timezone.utc)
utc7 = timezone(timedelta(hours=7))
now = now_utc.astimezone(utc7)
start_utc7 = start_dt.astimezone(utc7)
delta = start_dt - now_utc delta = start_dt - now_utc
total_sec = delta.total_seconds()
if delta.total_seconds() < 0: if total_sec < 0:
# Started already # Event is in the past
hours_ago = abs(delta.total_seconds()) / 3600 hours_ago = abs(total_sec) / 3600
if hours_ago < 1: if hours_ago < 1:
return "LIVE", 3 time_status = "LIVE"
time_urgency = 3
elif hours_ago < 4: elif hours_ago < 4:
return f"LIVE {int(hours_ago)}h", 3 time_status = f"LIVE {int(hours_ago)}h"
time_urgency = 3
elif hours_ago < 24: elif hours_ago < 24:
return f"Started {int(hours_ago)}h ago", 1 time_status = f"{int(hours_ago)}h ago"
time_urgency = 1
else: else:
days = int(hours_ago / 24) days = int(hours_ago / 24)
return f"{days}d ago", 0 time_status = f"{days}d ago"
time_urgency = 0
else: else:
# Starts in future # Event is in the future
hours_until = delta.total_seconds() / 3600 if total_sec < 3600:
if hours_until <= 0: mins = int(total_sec / 60)
return "LIVE", 3 time_status = f"In {mins}m"
elif hours_until < 1: time_urgency = 3
mins = int(delta.total_seconds() / 60) elif total_sec < 86400:
return f"In {mins}m", 3 hours_until = int(total_sec / 3600)
elif hours_until < 24: time_status = f"In {hours_until}h"
return f"In {int(hours_until)}h", 2 time_urgency = 2
else: else:
days = int(hours_until / 24) days = int(total_sec / 86400)
return f"In {days}d", 1 time_status = f"In {days}d"
except: time_urgency = 1
return "", 0
def get_match_time_str(e): abs_time = start_dt.astimezone(tz).strftime("%b %d, %H:%M ")
""" if tz == WIB:
Return just the time status string (e.g. 'LIVE', 'In 6h', 'In 1d'). abs_time += "WIB"
Uses startTime for actual match start time.
"""
start_str = e.get("startTime") or e.get("startDate", "")
if not start_str:
return "TBD"
try:
start_dt = datetime.fromisoformat(start_str.replace('Z', '+00:00'))
now_utc = datetime.now(timezone.utc)
delta = start_dt - now_utc
if delta.total_seconds() < 0:
hours_ago = abs(delta.total_seconds()) / 3600
if hours_ago < 1:
return "LIVE"
elif hours_ago < 4:
return f"LIVE {int(hours_ago)}h"
elif hours_ago < 24:
return f"{int(hours_ago)}h ago"
else:
days = int(hours_ago / 24)
return f"{days}d ago"
else: else:
hours_until = delta.total_seconds() / 3600 abs_time += start_dt.astimezone(tz).strftime("%Z")
if hours_until <= 0: return {"time_status": time_status, "time_urgency": time_urgency, "abs_time": abs_time}
return "LIVE" except Exception:
elif hours_until < 1: return {"time_status": "", "time_urgency": 0, "abs_time": "TBD"}
mins = int(delta.total_seconds() / 60)
return f"In {mins}m"
elif hours_until < 24:
return f"In {int(hours_until)}h"
else:
days = int(hours_until / 24)
return f"In {days}d"
except:
return ""
def filter_events(events, tradeable_only=True): def filter_events(events, tradeable_only=True):
""" """
@@ -325,6 +313,7 @@ def filter_events(events, tradeable_only=True):
return match_events, non_match_events return match_events, non_match_events
def sort_events(events): def sort_events(events):
return sorted(events, key=get_ml_volume, reverse=True) return sorted(events, key=get_ml_volume, reverse=True)
@@ -359,12 +348,12 @@ def format_event(e):
best_bid = float(ml.get("bestBid", 0)) if ml else 0 best_bid = float(ml.get("bestBid", 0)) if ml else 0
best_ask = float(ml.get("bestAsk", 0)) if ml else 0 best_ask = float(ml.get("bestAsk", 0)) if ml else 0
vol = get_ml_volume(e) vol = get_ml_volume(e)
time_status, urgency = get_match_time_status(e) td = _get_time_data(e)
return { return {
"title": e.get("title", ""), "title": e.get("title", ""),
"time_status": time_status, "time_status": td["time_status"],
"time_urgency": urgency, "time_urgency": td["time_urgency"],
"url": get_event_url(e), "url": get_event_url(e),
"livestream": e.get("resolutionSource"), "livestream": e.get("resolutionSource"),
"outcomes": outcomes, "outcomes": outcomes,
@@ -383,11 +372,12 @@ def format_detail_event(e):
] ]
active_markets = sorted(active_markets, key=lambda m: float(m.get("volume", 0)), reverse=True) active_markets = sorted(active_markets, key=lambda m: float(m.get("volume", 0)), reverse=True)
time_status, urgency = get_match_time_status(e) td = _get_time_data(e)
return { return {
"title": e.get("title", ""), "title": e.get("title", ""),
"time_status": time_status, "time_status": td["time_status"],
"abs_time": td["abs_time"],
"url": get_event_url(e), "url": get_event_url(e),
"livestream": e.get("resolutionSource"), "livestream": e.get("resolutionSource"),
"outcomes": json.loads(ml.get("outcomes", "[]")) if ml else [], "outcomes": json.loads(ml.get("outcomes", "[]")) if ml else [],
@@ -414,48 +404,6 @@ def format_detail_event(e):
# DISPLAY # DISPLAY
# ============================================================ # ============================================================
def get_start_time_wib(e):
"""Return (date_time_str, relative_str) for display."""
start_str = e.get("startTime") or e.get("startDate", "")
if not start_str:
return "TBD", ""
try:
start_dt = datetime.fromisoformat(start_str.replace('Z', '+00:00'))
now_utc = datetime.now(timezone.utc)
utc7 = timezone(timedelta(hours=7))
start_utc7 = start_dt.astimezone(utc7)
# Absolute: "Mar 25, 19:00 WIB"
abs_str = start_utc7.strftime("%b %d, %H:%M WIB")
# Relative: "In 5h", "In 10h", "LIVE", etc.
delta = start_dt - now_utc
if delta.total_seconds() < 0:
hours_ago = abs(delta.total_seconds()) / 3600
if hours_ago < 1:
rel_str = "LIVE"
elif hours_ago < 24:
rel_str = f"{int(hours_ago)}h ago"
else:
days = int(hours_ago / 24)
rel_str = f"{days}d ago"
else:
hours_until = delta.total_seconds() / 3600
if hours_until <= 0:
rel_str = "LIVE"
elif hours_until < 1:
mins_until = int(delta.total_seconds() / 60)
rel_str = f"In {mins_until}m"
elif hours_until < 24:
rel_str = f"In {int(hours_until)}h"
else:
days = int(hours_until / 24)
rel_str = f"In {days}d"
return abs_str, rel_str
except:
return "TBD", ""
def get_header_date(): def get_header_date():
"""Return current date string like 'Mar 25, 2026'""" """Return current date string like 'Mar 25, 2026'"""
now_utc = datetime.now(timezone.utc) now_utc = datetime.now(timezone.utc)
@@ -511,7 +459,9 @@ def print_browse(match_events, non_match_events, category, total_raw, total_fetc
vol = f["volume"] vol = f["volume"]
title = f["title"] title = f["title"]
url = f["url"] url = f["url"]
start_time_wib, rel_time = get_start_time_wib(e) td = _get_time_data(e)
start_time_wib = td["abs_time"]
rel_time = td["time_status"]
team_a = outcomes[0] if len(outcomes) > 0 else "?" team_a = outcomes[0] if len(outcomes) > 0 else "?"
team_b = outcomes[1] if len(outcomes) > 1 else "?" team_b = outcomes[1] if len(outcomes) > 1 else "?"
@@ -539,7 +489,9 @@ def print_browse(match_events, non_match_events, category, total_raw, total_fetc
for i, e in enumerate(non_match_events[:non_matches_max], 1): for i, e in enumerate(non_match_events[:non_matches_max], 1):
title = e.get("title", "?") title = e.get("title", "?")
url = get_event_url(e) url = get_event_url(e)
start_time_wib, rel_time = get_start_time_wib(e) td = _get_time_data(e)
start_time_wib = td["abs_time"]
rel_time = td["time_status"]
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", []))
@@ -549,17 +501,11 @@ def print_browse(match_events, non_match_events, category, total_raw, total_fetc
print(f" Markets: {market_count} | Total Vol: ${total_vol:,.0f}") print(f" Markets: {market_count} | Total Vol: ${total_vol:,.0f}")
def print_detail(e, detail): def print_detail(e, detail):
from datetime import datetime, timezone, timedelta
now_utc = datetime.now(timezone.utc)
utc7 = timezone(timedelta(hours=7))
now_utc7 = now_utc.astimezone(utc7)
print(f"\n{detail['title']}") print(f"\n{detail['title']}")
print(f"URL: {detail['url']}") print(f"URL: {detail['url']}")
print(f"Livestream: {detail['livestream']}") print(f"Livestream: {detail['livestream']}")
spread_str = format_spread(detail["best_bid"], detail["best_ask"]) if detail["best_bid"] and detail["best_ask"] else "N/A" spread_str = format_spread(detail["best_bid"], detail["best_ask"]) if detail["best_bid"] and detail["best_ask"] else "N/A"
time_str = get_match_time_str(e)
print(f"\n{detail['time_status']}") print(f"\n{detail['time_status']}")
print(f"ML: {detail['outcomes'][0]} {format_odds(float(detail['prices'][0]))} vs {detail['outcomes'][1]} {format_odds(float(detail['prices'][1]))}") print(f"ML: {detail['outcomes'][0]} {format_odds(float(detail['prices'][0]))} vs {detail['outcomes'][1]} {format_odds(float(detail['prices'][1]))}")
print(f"ML Vol: ${detail['volume']:,.0f} | {spread_str}") print(f"ML Vol: ${detail['volume']:,.0f} | {spread_str}")
@@ -576,14 +522,44 @@ def print_detail(e, detail):
# TELEGRAM # TELEGRAM
# ============================================================ # ============================================================
def escape_html(text):
"""Escape HTML-sensitive characters for Telegram parse_mode=HTML."""
return (text
.replace("&", "&amp;")
.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): def send_to_telegram(match_events, non_match_events, category, matches_only=False, non_matches_only=False):
"""Send browse results to Telegram. Reads BOT_TOKEN and CHAT_ID from environment.""" """Send browse results to Telegram. Reads TELEGRAM_BOT_TOKEN and CHAT_ID from environment."""
import os import os
bot_token = os.environ.get("BOT_TOKEN") bot_token = os.environ.get("TELEGRAM_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:
print("WARNING: BOT_TOKEN or CHAT_ID not set in environment. Skipping Telegram send.") raise RuntimeError("TELEGRAM_BOT_TOKEN or CHAT_ID not set in environment")
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)
@@ -596,19 +572,8 @@ 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):
result = subprocess.run( msg_id = send_telegram_message(bot_token, chat_id, text)
["curl", "-s", f"https://api.telegram.org/bot{bot_token}/sendMessage", print(f" Sent msg {msg_id}")
"-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}"]
@@ -627,14 +592,16 @@ def send_to_telegram(match_events, non_match_events, category, matches_only=Fals
vol = get_ml_volume(e) vol = get_ml_volume(e)
title = e.get("title", "?") title = e.get("title", "?")
url = get_event_url(e) url = get_event_url(e)
start_time_wib, rel_time = get_start_time_wib(e) td = _get_time_data(e)
start_time_wib = td["abs_time"]
rel_time = td["time_status"]
team_a = outcomes[0] if len(outcomes) > 0 else "?" team_a = outcomes[0] if len(outcomes) > 0 else "?"
team_b = outcomes[1] if len(outcomes) > 1 else "?" team_b = outcomes[1] if len(outcomes) > 1 else "?"
odds_a = format_odds(float(prices[0])) if len(prices) > 0 else "?" odds_a = format_odds(float(prices[0])) if len(prices) > 0 else "?"
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}\">{title_clean}</a>") lines.append(f"<b>{i}.</b> <a href=\"{url}\">{escape_html(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:
@@ -652,10 +619,12 @@ def send_to_telegram(match_events, non_match_events, category, matches_only=Fals
for i, e in enumerate(non_match_events, 1): for i, e in enumerate(non_match_events, 1):
title = e.get("title", "?") title = e.get("title", "?")
url = get_event_url(e) url = get_event_url(e)
start_time_wib, rel_time = get_start_time_wib(e) td = _get_time_data(e)
start_time_wib = td["abs_time"]
rel_time = td["time_status"]
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}\">{title}</a>") lines.append(f"<b>{i}.</b> <a href=\"{url}\">{escape_html(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("")
@@ -737,7 +706,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 (BOT_TOKEN and CHAT_ID must be set in environment).") help="Send results to Telegram (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:

View File

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

View File

@@ -0,0 +1,324 @@
"""
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
from datetime import datetime, timezone, timedelta
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")
class TestTimeFunctions(unittest.TestCase):
"""Tests for _get_time_data() unified helper.
These tests verify the helper returns correct time_status, time_urgency,
and abs_time for various event scenarios. Callers extract the fields they
need from the returned dict.
"""
def _make_event(self, start_time):
"""Helper to create a minimal event with a startTime."""
return {"startTime": start_time}
def _frozen_dt(self, year, month, day, hour, minute, second=0):
return datetime(year, month, day, hour, minute, second,
tzinfo=timezone.utc)
def _mock_datetime(self, frozen):
"""Return a mock datetime class that freezes now() to the given datetime."""
class MockDatetime:
@staticmethod
def now(tz=None):
if tz is None:
return frozen
return frozen.astimezone(tz)
fromisoformat = staticmethod(datetime.fromisoformat)
def __call__(self, *a, **k):
return datetime(*a, **k)
return MockDatetime
# === _get_time_data core tests ===
def test_get_time_data_tbd(self):
"""No startTime -> TBD/0urgency/abs TBD."""
from browse import _get_time_data
td = _get_time_data({})
self.assertEqual(td["time_status"], "TBD")
self.assertEqual(td["time_urgency"], 0)
self.assertEqual(td["abs_time"], "TBD")
def test_get_time_data_in_30m(self):
"""Starts in 30 minutes -> 'In 30m', urgency 3."""
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
td = _get_time_data(self._make_event("2026-03-25T12:30:00Z"))
self.assertEqual(td["time_status"], "In 30m")
self.assertEqual(td["time_urgency"], 3)
self.assertIn("WIB", td["abs_time"])
def test_get_time_data_in_6h(self):
"""Starts in 6 hours -> 'In 6h', urgency 2."""
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
td = _get_time_data(self._make_event("2026-03-25T18:00:00Z"))
self.assertEqual(td["time_status"], "In 6h")
self.assertEqual(td["time_urgency"], 2)
self.assertIn("WIB", td["abs_time"])
def test_get_time_data_in_2d(self):
"""Starts in 2 days -> 'In 2d', urgency 1."""
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
td = _get_time_data(self._make_event("2026-03-27T12:00:00Z"))
self.assertEqual(td["time_status"], "In 2d")
self.assertEqual(td["time_urgency"], 1)
def test_get_time_data_live(self):
"""Started 30 minutes ago -> 'LIVE', urgency 3."""
frozen = self._frozen_dt(2026, 3, 25, 12, 30, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
self.assertEqual(td["time_status"], "LIVE")
self.assertEqual(td["time_urgency"], 3)
self.assertIn("WIB", td["abs_time"])
def test_get_time_data_started_2h_ago(self):
"""Started 2 hours ago -> 'LIVE 2h', urgency 3."""
frozen = self._frozen_dt(2026, 3, 25, 14, 0, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
self.assertEqual(td["time_status"], "LIVE 2h")
self.assertEqual(td["time_urgency"], 3)
def test_get_time_data_started_12h_ago(self):
"""Started 12 hours ago -> '12h ago', urgency 1."""
frozen = self._frozen_dt(2026, 3, 26, 0, 0, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
self.assertEqual(td["time_status"], "12h ago")
self.assertEqual(td["time_urgency"], 1)
def test_get_time_data_started_2d_ago(self):
"""Started 2 days ago -> '2d ago', urgency 0."""
frozen = self._frozen_dt(2026, 3, 27, 12, 0, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
td = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
self.assertEqual(td["time_status"], "2d ago")
self.assertEqual(td["time_urgency"], 0)
def test_get_time_data_abs_time_format(self):
"""abs_time is formatted correctly in WIB."""
frozen = self._frozen_dt(2026, 3, 25, 12, 0, 0)
with patch('browse.datetime', self._mock_datetime(frozen)):
from browse import _get_time_data
# 19:00 UTC = 02:00 WIB next day
td = _get_time_data(self._make_event("2026-03-26T02:00:00Z"))
self.assertIn("WIB", td["abs_time"])
# UTC 12:00 -> WIB 19:00 same day
td2 = _get_time_data(self._make_event("2026-03-25T12:00:00Z"))
self.assertEqual(td2["abs_time"], "Mar 25, 19:00 WIB")
if __name__ == "__main__":
unittest.main()