Skip to content

2026 01 01 qegl a79ac #16

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
roto31 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

2026 01 01 qegl a79ac #16

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bce5134
Enhance logo URL resolution in IPTV API to validate channel logos aga…
roto31 Jan 2, 2026
be3d98f
Refactor channel icon handling to use channel numbers for consistency…
roto31 Jan 2, 2026
2ead1f3
Remove Rust port files (ErsatzTV-Windows reference)
roto31 Jan 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 35 additions & 6 deletions streamtv/api/auth.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,21 +201,35 @@ async def youtube_set_cookies(request: Request, file: UploadFile = File(...)):
cookies_dir = Path("data/cookies")
cookies_dir.mkdir(parents=True, exist_ok=True)

# Save uploaded file
cookies_path = cookies_dir / "youtube_cookies.txt"
# Save uploaded file with site name format (youtube.com_cookies.txt or www.youtube.com_cookies.txt)
# Try both formats - prefer youtube.com (without www.) as it's more common
cookies_path_new1 = cookies_dir / "youtube.com_cookies.txt"
cookies_path_new2 = cookies_dir / "www.youtube.com_cookies.txt"
cookies_path_old = cookies_dir / "youtube_cookies.txt"
cookies_path = cookies_path_new1 # Use youtube.com format by default

try:
with open(cookies_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)

# Also save to old format for backward compatibility
try:
with open(cookies_path_old, "wb") as buffer:
file.file.seek(0) # Reset file pointer
shutil.copyfileobj(file.file, buffer)
except Exception:
pass # Non-critical if old format save fails

# Validate it's a valid cookies file (basic check)
with open(cookies_path, "r") as f:
content = f.read()
if "youtube.com" not in content.lower() and "# Netscape" not in content:
cookies_path.unlink() # Delete invalid file
if cookies_path_old.exists():
cookies_path_old.unlink()
raise HTTPException(status_code=400, detail="Invalid cookies file format")

# Update config and persist to file
# Update config and persist to file (use new format)
config.update_section("youtube", {
"cookies_file": str(cookies_path.absolute()),
"use_authentication": True
Expand All @@ -233,6 +247,8 @@ async def youtube_set_cookies(request: Request, file: UploadFile = File(...)):
logger.error(f"Error uploading cookies file: {e}")
if cookies_path.exists():
cookies_path.unlink()
if cookies_path_old.exists():
cookies_path_old.unlink()
raise HTTPException(status_code=500, detail=str(e))


Expand All @@ -258,21 +274,34 @@ async def archive_set_cookies(request: Request, file: UploadFile = File(...)):
cookies_dir = Path("data/cookies")
cookies_dir.mkdir(parents=True, exist_ok=True)

# Save uploaded file
cookies_path = cookies_dir / "archive_cookies.txt"
# Save uploaded file with site name format (archive.org_cookies.txt)
# Also keep old format for backward compatibility
cookies_path_new = cookies_dir / "archive.org_cookies.txt"
cookies_path_old = cookies_dir / "archive_cookies.txt"
cookies_path = cookies_path_new # Use new format by default

try:
with open(cookies_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)

# Also save to old format for backward compatibility
try:
with open(cookies_path_old, "wb") as buffer:
file.file.seek(0) # Reset file pointer
shutil.copyfileobj(file.file, buffer)
except Exception:
pass # Non-critical if old format save fails

# Validate it's a valid cookies file (basic check)
with open(cookies_path, "r") as f:
content = f.read()
if "archive.org" not in content.lower() and "# Netscape" not in content:
cookies_path.unlink() # Delete invalid file
if cookies_path_old.exists():
cookies_path_old.unlink()
raise HTTPException(status_code=400, detail="Invalid cookies file format. Must contain archive.org cookies.")

# Update config and persist to file
# Update config and persist to file (use new format)
config.update_section("archive_org", {
"cookies_file": str(cookies_path.absolute()),
"use_authentication": True
Expand Down
27 changes: 22 additions & 5 deletions streamtv/api/channels.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,22 +368,33 @@
icons_dir = project_root / "data" / "channel_icons"
icons_dir.mkdir(parents=True, exist_ok=True)

# Generate filename: channel_{channel_id}.png
icon_filename = f"channel_{channel_id}.png"
# Generate filename using channel NUMBER (not database ID) for consistency with XMLTV/HDHomeRun
# This ensures icons match channel numbers used in lineup.json and XMLTV channel IDs
icon_filename = f"channel_{channel.number}.png"
icon_path = icons_dir / icon_filename

# If there's an old icon file using database ID, remove it
old_icon_filename = f"channel_{channel_id}.png"
old_icon_path = icons_dir / old_icon_filename
if old_icon_path.exists() and old_icon_path != icon_path:

Check failure

Code scanning / CodeQL

Uncontrolled data used in path expression High

This path depends on a
user-provided value
Loading
.

Copilot Autofix

AI about 1 month ago

Copilot could not generate an autofix suggestion

Copilot could not generate an autofix suggestion for this alert. Try pushing a new commit or if the problem persists contact support.

try:
old_icon_path.unlink()

Check failure

Code scanning / CodeQL

Uncontrolled data used in path expression High

This path depends on a
user-provided value
Loading
.

Copilot Autofix

AI about 1 month ago

General fix approach: Ensure that any value derived from user input and used in a filesystem path is validated or constrained so it cannot cause access outside the intended directory or unexpected file operations. Typical strategies are to validate/normalize paths or to strictly validate identifiers used inside filenames.

Best fix here without changing behavior: The only tainted piece is channel_id, which is already documented and used as an integer primary key. In FastAPI, an untyped path parameter could be any string, but here the parameter is annotated as int. To make this explicit and satisfy both security and the analyzer, we can add a small runtime validation at the top of upload_channel_icon to ensure channel_id is a positive integer. This prevents odd or negative values from being used in filenames, tightly constrains the space of possible filenames to channel_<positive-int>.png, and does not alter normal behavior for valid requests. We keep all path construction logic the same.

Concrete changes (all in streamtv/api/channels.py):

  • In upload_channel_icon, right after the function signature (before querying the database), add a check:

    if channel_id <= 0:
    raise HTTPException(status_code=400, detail="Invalid channel ID")

    This guarantees that channel_id is a simple positive integer, so old_icon_filename and icon_filename remain within a predictable, safe pattern.

No new imports are needed; HTTPException and status are already imported.

Suggested changeset 1
streamtv/api/channels.py

Autofix patch

Autofix patch
Run the following command in your local git repository to apply this patch
cat << 'EOF' | git apply
diff --git a/streamtv/api/channels.py b/streamtv/api/channels.py
--- a/streamtv/api/channels.py
+++ b/streamtv/api/channels.py
@@ -345,6 +345,10 @@
     db: Session = Depends(get_db)
 ):
     """Upload a PNG icon for a channel"""
+    # Validate channel ID is a positive integer
+    if channel_id <= 0:
+        raise HTTPException(status_code=400, detail="Invalid channel ID")
+
     # Validate channel exists
     channel = db.query(Channel).filter(Channel.id == channel_id).first()
     if not channel:
EOF
@@ -345,6 +345,10 @@
db: Session = Depends(get_db)
):
"""Upload a PNG icon for a channel"""
# Validate channel ID is a positive integer
if channel_id <= 0:
raise HTTPException(status_code=400, detail="Invalid channel ID")

# Validate channel exists
channel = db.query(Channel).filter(Channel.id == channel_id).first()
if not channel:
Copilot is powered by AI and may make mistakes. Always verify output.
logger.info(f"Removed old icon file using database ID: {old_icon_filename}")
except Exception as e:
logger.warning(f"Could not remove old icon file {old_icon_filename}: {e}")

try:
# Save the uploaded file
with open(icon_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)

# Update channel logo_path to point to the static file
# Update channel logo_path to point to the static file (using channel number)
logo_url = f"/static/channel_icons/{icon_filename}"
channel.logo_path = logo_url
db.commit()
db.refresh(channel)

logger.info(f"Uploaded icon for channel {channel_id} ({channel.name}): {icon_path}")
logger.info(f"Uploaded icon for channel {channel.number} (ID: {channel_id}, {channel.name}): {icon_path}")

return channel
except Exception as e:
Expand All @@ -410,8 +421,14 @@
# Determine icons directory
project_root = Path(__file__).parent.parent.parent
icons_dir = project_root / "data" / "channel_icons"
icon_filename = f"channel_{channel_id}.png"

# Try channel number first (preferred), then fallback to database ID for backward compatibility
icon_filename = f"channel_{channel.number}.png"
icon_path = icons_dir / icon_filename
if not icon_path.exists():
# Fallback to database ID for old icons
icon_filename = f"channel_{channel_id}.png"
icon_path = icons_dir / icon_filename

# Delete the file if it exists
if icon_path.exists():
Expand Down
71 changes: 67 additions & 4 deletions streamtv/api/iptv.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,16 +33,46 @@ def _xml(value) -> str:
def _resolve_logo_url(channel, base_url: str) -> Optional[str]:
"""
Build an absolute logo URL for M3U/XMLTV.
- Uses channel.logo_path if provided.
- Uses channel.logo_path if provided and it matches the channel number.
- Falls back to /static/channel_icons/channel_<number>.png.

Note: Some channels have incorrect logo_path values using database IDs instead of channel numbers.
We validate and use the correct path based on channel number.
"""
logo_path = channel.logo_path
if logo_path:
# If it's an external URL, use it directly
if logo_path.startswith('http'):
return logo_path
if logo_path.startswith('/'):
return f"{base_url}{logo_path}"
return f"{base_url}/{logo_path}"

# Check if logo_path contains a channel number that matches this channel
# Extract any number from the logo_path filename
import re
logo_filename = logo_path.split('/')[-1] # Get just the filename
logo_match = re.search(r'channel_(\d+)\.png', logo_filename)

if logo_match:
logo_number = logo_match.group(1)
channel_number_str = str(channel.number)
# If the logo path number matches the channel number, use it
if logo_number == channel_number_str:
if logo_path.startswith('/'):
return f"{base_url}{logo_path}"
return f"{base_url}/{logo_path}"
else:
# Logo path has wrong number (likely database ID), use fallback
logger.debug(f"Channel {channel.number}: logo_path '{logo_path}' has number {logo_number} (doesn't match channel number), using fallback")
else:
# No number found in logo_path, might be a custom path - use it if it's a static path
if '/static/channel_icons/' in logo_path or '/channel_icons/' in logo_path:
if logo_path.startswith('/'):
return f"{base_url}{logo_path}"
return f"{base_url}/{logo_path}"
# Custom path not in channel_icons - use it as-is
if logo_path.startswith('/'):
return f"{base_url}{logo_path}"
return f"{base_url}/{logo_path}"

# Default fallback based on channel number icon
return f"{base_url}/static/channel_icons/channel_{channel.number}.png"

Expand Down Expand Up @@ -308,9 +338,25 @@ async def get_epg(
plex_client = None

# Channel definitions - ensure Plex-compatible format
# #region agent log
import json
try:
with open('/Users/roto1231/Documents/XCode Projects/StreamTV/.cursor/debug.log', 'a') as f:
f.write(json.dumps({"sessionId":"debug-session","runId":"run1","hypothesisId":"A","location":"api/iptv.py:311","message":"XMLTV: Starting channel definitions","data":{"channel_count":len(channels),"base_url":base_url},"timestamp":int(__import__('time').time()*1000)})+'\n')
except: pass
# #endregion

for channel in channels:
# Use channel number as ID (Plex expects numeric or alphanumeric IDs)
channel_id = str(channel.number).strip()

# #region agent log
try:
with open('/Users/roto1231/Documents/XCode Projects/StreamTV/.cursor/debug.log', 'a') as f:
f.write(json.dumps({"sessionId":"debug-session","runId":"run1","hypothesisId":"A","location":"api/iptv.py:316","message":"XMLTV: Channel ID generated","data":{"channel_number":channel.number,"channel_id":channel_id,"channel_id_type":type(channel_id).__name__,"channel_id_repr":repr(channel_id),"channel_name":channel.name},"timestamp":int(__import__('time').time()*1000)})+'\n')
except: pass
# #endregion

xml_content += f' <channel id="{_xml(channel_id)}">\n'

# Primary display name (required)
Expand All @@ -325,6 +371,14 @@ async def get_epg(

# Logo/icon (Plex expects absolute URLs). Fall back to default icon by number.
logo_url = _resolve_logo_url(channel, base_url)

# #region agent log
try:
with open('/Users/roto1231/Documents/XCode Projects/StreamTV/.cursor/debug.log', 'a') as f:
f.write(json.dumps({"sessionId":"debug-session","runId":"run1","hypothesisId":"D","location":"api/iptv.py:332","message":"XMLTV: Icon URL resolved","data":{"channel_number":channel.number,"logo_url":logo_url,"logo_path":getattr(channel,'logo_path',None)},"timestamp":int(__import__('time').time()*1000)})+'\n')
except: pass
# #endregion

if logo_url:
xml_content += f' <icon src="{_xml(logo_url)}"/>\n'

Expand Down Expand Up @@ -884,6 +938,14 @@ async def get_epg(
generation_time = time.time() - perf_start_time
logger.info(f"XMLTV EPG generated in {generation_time:.2f}s ({len(xml_content)} bytes)")

# #region agent log
try:
import json
with open('/Users/roto1231/Documents/XCode Projects/StreamTV/.cursor/debug.log', 'a') as f:
f.write(json.dumps({"sessionId":"debug-session","runId":"run1","hypothesisId":"B","location":"api/iptv.py:887","message":"XMLTV: Response prepared","data":{"content_length":len(xml_content),"media_type":"application/xml; charset=utf-8","generation_time":generation_time},"timestamp":int(__import__('time').time()*1000)})+'\n')
except: pass
# #endregion

return Response(
content=xml_content,
media_type="application/xml; charset=utf-8",
Expand Down Expand Up @@ -1573,3 +1635,4 @@ async def generate():
except Exception as e:
logger.error(f"Unexpected error streaming media {media_id}: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

Loading
Loading