fix(logos): harden the plugin logo download and share core HTTP headers

download_missing_logo / LogoDownloader.download_logo, the path the
scoreboard plugins use, read response.content with no size cap and wrote
straight to the final path, so a failed or corrupt download could be left
in place and cached as the logo. It now goes through fetch_logo: streamed
with a 10 MB cap, image/* only, decoded by Pillow, converted to RGBA once,
and moved into place atomically. A failure leaves no partial or temp file
and keeps any logo already on disk. LogoHelper._download_logo delegates to
the same code. Public signatures and return values are unchanged; saved
files are pixel-identical to before (RGBA, palette+tRNS, L+tRNS, LA, JPEG).

download_missing_logo reuses one downloader per thread instead of a new
Session per logo. Per thread rather than behind a lock: Session is not
documented thread-safe, and a lock would serialise every plugin's
downloads behind the slowest one.

Placeholders are written atomically, without the test_write.tmp probe.

The logo downloader and background data service now send the real
ChuckBuilds User-Agent from src.common.api_helper (USER_AGENT,
DEFAULT_HTTP_HEADERS) instead of a yourusername/contact@example.com
placeholder, and no longer hand-set Accept-Encoding: br (brotli is not
installed).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-22 17:17:10 -04:00
co-authored by Claude Opus 5.5
parent 967f3a0567
commit 2e849d7de7
8 changed files with 611 additions and 150 deletions
+20 -4
View File
@@ -8,14 +8,32 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins.
import logging
import time
from datetime import datetime
from types import MappingProxyType
from src.common.espn_dates import ESPN_MAX_LIMIT
from typing import Any, Dict, Optional
from typing import Any, Dict, Mapping, Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
#: The User-Agent core sends to ESPN and other data APIs. It names the client
#: and links to it: around 2026-08-04 ESPN began 403ing bare custom tokens
#: (and browser strings), and this form is what it accepts.
USER_AGENT = 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)'
#: Base headers for core's JSON API requests. Read-only; pass
#: ``{**DEFAULT_HTTP_HEADERS, ...}`` to add to it. There is deliberately no
#: Accept-Encoding: requests advertises only what urllib3 can decode here
#: (``br`` needs the optional brotli package, which is not a requirement), so a
#: hand-set ``br`` invites a body the client cannot read.
DEFAULT_HTTP_HEADERS: Mapping[str, str] = MappingProxyType({
'User-Agent': USER_AGENT,
'Accept': 'application/json',
'Accept-Language': 'en-US,en;q=0.9',
})
class APIHelper:
"""
Helper class for HTTP requests, caching, and ESPN API integration.
@@ -57,9 +75,7 @@ class APIHelper:
# Default headers
self.session.headers.update({
# Identifies the client and links to it: ESPN began 403ing bare
# custom tokens (and browser strings) around 2026-08-04.
'User-Agent': 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)',
'User-Agent': USER_AGENT,
'Accept': 'application/json',
'Accept-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
+13 -57
View File
@@ -6,8 +6,6 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins.
"""
import logging
import os
import tempfile
import time
from pathlib import Path
from typing import Dict, List, Optional, Union
@@ -16,9 +14,7 @@ import requests
from PIL import Image
from src.common.permission_utils import (
ensure_directory_permissions,
ensure_file_permissions,
get_assets_dir_mode,
get_assets_file_mode
)
# How long a missing logo stays remembered as missing.
@@ -61,6 +57,7 @@ def _usable_scale(scale) -> float:
# Well above any real team logo; bounds what a remote URL can write to disk.
# The cap for every logo download: src.logo_downloader.fetch_logo uses it too.
MAX_LOGO_BYTES = 10 * 1024 * 1024
@@ -408,64 +405,23 @@ class LogoHelper:
self._cache_order.append(cache_key)
def _download_logo(self, url: str, file_path: Path) -> None:
"""Download logo from URL.
"""Download a logo from ``url`` to ``file_path``; raises on failure.
The response size is capped and the saved file is verified as a
decodable image before it is left on disk: a logo URL is remote
input, and without this an oversized or malformed response would
be cached for every later load_logo() call to trip over.
Delegates to ``src.logo_downloader.fetch_logo``, the same hardened
download the scoreboard plugins use: streamed and capped at
``MAX_LOGO_BYTES``, ``image/*`` only, decoded by Pillow, stored as an
RGBA PNG, and moved into place atomically -- a failure leaves neither
a partial file nor a temp file. Uses this helper's own session.
The body is streamed and counted as it arrives rather than read
through response.content, which buffers the whole thing first —
a server that omits Content-Length and never stops sending would
exhaust memory before any size check could run. Nothing lands at
file_path until the download completes and decodes, so a failed
download cannot leave a truncated logo behind either.
Imported lazily: src.logo_downloader imports src.common, so a
module-level import here would be circular.
"""
from src.logo_downloader import fetch_logo
# Ensure directory exists with proper permissions
ensure_directory_permissions(file_path.parent, get_assets_dir_mode())
# A unique temp name, not a fixed "<name>.part": two plugins can
# ask for the same logo at once, and a shared name would let them
# interleave writes into one file, publish the mixture, or delete
# each other's partial. Same directory, so os.replace stays atomic.
fd, tmp_name = tempfile.mkstemp(
dir=str(file_path.parent), prefix=file_path.name + '.', suffix='.part')
tmp_path = Path(tmp_name)
try:
# fdopen outermost so the descriptor mkstemp handed back is
# always adopted and closed, including when the request itself
# raises — load_logo_with_download swallows that, so a leak
# here would accumulate quietly on a URL that keeps failing.
with os.fdopen(fd, 'wb') as f:
with self.session.get(url, timeout=30, stream=True) as response:
response.raise_for_status()
downloaded = 0
for chunk in response.iter_content(chunk_size=64 * 1024):
if not chunk:
continue
downloaded += len(chunk)
if downloaded > MAX_LOGO_BYTES:
raise ValueError(
f"Logo at {url} exceeds the "
f"{MAX_LOGO_BYTES}-byte limit; not saved")
f.write(chunk)
# Verify it decodes before it becomes the cached logo. PIL
# raises DecompressionBombError past its own pixel limit; a
# partial or non-image response raises UnidentifiedImageError
# (an OSError subclass).
with Image.open(tmp_path) as probe:
probe.load()
os.replace(tmp_path, file_path)
except BaseException:
tmp_path.unlink(missing_ok=True)
raise
# Set proper file permissions after saving
ensure_file_permissions(file_path, get_assets_file_mode())
fetch_logo(self.session, url, file_path, timeout=30,
max_bytes=MAX_LOGO_BYTES)
self.logger.debug(f"Downloaded logo to {file_path}")
def _create_placeholder_logo(self, team_abbr: str,