mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-17 16:50:53 +00:00
fix: guard CoreSettings list getters against corrupted stored data
- Sanitize series_rules on read (get_dvr_series_rules) and write (set_dvr_series_rules), filtering non-dict elements and non-list values - Sanitize series_rules in CoreSettingsSerializer.update() to cover the generic settings API path (Settings page round-trip) - Harden EPG ignore list getters (prefixes, suffixes, custom) with shared _safe_string_list helper to filter non-string/non-list values - Add 13 unit tests covering all corruption scenarios
This commit is contained in:
None
parent
5b9b0162da
commit
0d6979ce1d
3 changed files with 179 additions and 8 deletions
|
|
@ -239,17 +239,24 @@ class CoreSettings(models.Model):
|
|||
"epg_match_ignore_custom": [],
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def _safe_string_list(cls, value):
|
||||
"""Return a list of strings, filtering out non-list or non-string values."""
|
||||
if not isinstance(value, list):
|
||||
return []
|
||||
return [v for v in value if isinstance(v, str)]
|
||||
|
||||
@classmethod
|
||||
def get_epg_match_ignore_prefixes(cls):
|
||||
return cls.get_epg_settings().get("epg_match_ignore_prefixes", [])
|
||||
return cls._safe_string_list(cls.get_epg_settings().get("epg_match_ignore_prefixes", []))
|
||||
|
||||
@classmethod
|
||||
def get_epg_match_ignore_suffixes(cls):
|
||||
return cls.get_epg_settings().get("epg_match_ignore_suffixes", [])
|
||||
return cls._safe_string_list(cls.get_epg_settings().get("epg_match_ignore_suffixes", []))
|
||||
|
||||
@classmethod
|
||||
def get_epg_match_ignore_custom(cls):
|
||||
return cls.get_epg_settings().get("epg_match_ignore_custom", [])
|
||||
return cls._safe_string_list(cls.get_epg_settings().get("epg_match_ignore_custom", []))
|
||||
|
||||
# DVR Settings
|
||||
@classmethod
|
||||
|
|
@ -312,12 +319,16 @@ class CoreSettings(models.Model):
|
|||
|
||||
@classmethod
|
||||
def get_dvr_series_rules(cls):
|
||||
return cls.get_dvr_settings().get("series_rules", [])
|
||||
rules = cls.get_dvr_settings().get("series_rules", [])
|
||||
if not isinstance(rules, list):
|
||||
return []
|
||||
return [r for r in rules if isinstance(r, dict)]
|
||||
|
||||
@classmethod
|
||||
def set_dvr_series_rules(cls, rules):
|
||||
cls._update_group(DVR_SETTINGS_KEY, "DVR Settings", {"series_rules": rules})
|
||||
return rules
|
||||
clean = [r for r in rules if isinstance(r, dict)] if isinstance(rules, list) else []
|
||||
cls._update_group(DVR_SETTINGS_KEY, "DVR Settings", {"series_rules": clean})
|
||||
return clean
|
||||
|
||||
# Proxy Settings
|
||||
@classmethod
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import json
|
|||
import ipaddress
|
||||
|
||||
from rest_framework import serializers
|
||||
from .models import CoreSettings, UserAgent, StreamProfile, NETWORK_ACCESS_KEY
|
||||
from .models import CoreSettings, UserAgent, StreamProfile, DVR_SETTINGS_KEY, NETWORK_ACCESS_KEY
|
||||
|
||||
|
||||
class UserAgentSerializer(serializers.ModelSerializer):
|
||||
|
|
@ -64,6 +64,19 @@ class CoreSettingsSerializer(serializers.ModelSerializer):
|
|||
}
|
||||
)
|
||||
|
||||
# Sanitize series_rules when DVR settings are saved through the
|
||||
# generic settings API (e.g. Settings page round-trip) to prevent
|
||||
# corrupted non-dict entries from persisting.
|
||||
if instance.key == DVR_SETTINGS_KEY:
|
||||
value = validated_data.get("value")
|
||||
if isinstance(value, dict) and "series_rules" in value:
|
||||
rules = value["series_rules"]
|
||||
value["series_rules"] = (
|
||||
[r for r in rules if isinstance(r, dict)]
|
||||
if isinstance(rules, list)
|
||||
else []
|
||||
)
|
||||
|
||||
result = super().update(instance, validated_data)
|
||||
|
||||
# Note: Cache invalidation and notification sync is handled by post_save signal
|
||||
|
|
|
|||
149
core/tests.py
149
core/tests.py
|
|
@ -1,3 +1,150 @@
|
|||
from django.test import TestCase
|
||||
|
||||
# Create your tests here.
|
||||
from core.models import CoreSettings, DVR_SETTINGS_KEY, EPG_SETTINGS_KEY
|
||||
|
||||
|
||||
class GetDvrSeriesRulesTest(TestCase):
|
||||
"""Verify get_dvr_series_rules handles corrupted stored data."""
|
||||
|
||||
def _set_series_rules_raw(self, raw_value):
|
||||
"""Write a raw series_rules value into the DB, bypassing set_dvr_series_rules."""
|
||||
obj, _ = CoreSettings.objects.get_or_create(
|
||||
key=DVR_SETTINGS_KEY,
|
||||
defaults={"name": "DVR Settings", "value": {}},
|
||||
)
|
||||
current = obj.value if isinstance(obj.value, dict) else {}
|
||||
current["series_rules"] = raw_value
|
||||
obj.value = current
|
||||
obj.save()
|
||||
|
||||
def test_valid_rules_returned_as_is(self):
|
||||
rules = [{"tvg_id": "abc", "mode": "all", "title": "Show"}]
|
||||
self._set_series_rules_raw(rules)
|
||||
result = CoreSettings.get_dvr_series_rules()
|
||||
self.assertEqual(result, rules)
|
||||
|
||||
def test_non_dict_elements_filtered(self):
|
||||
"""Strings in the list cause 'str' has no attribute 'get'."""
|
||||
self._set_series_rules_raw(["bad_string", {"tvg_id": "abc", "mode": "all", "title": ""}])
|
||||
result = CoreSettings.get_dvr_series_rules()
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertEqual(result[0]["tvg_id"], "abc")
|
||||
|
||||
def test_non_list_value_returns_empty(self):
|
||||
"""If series_rules is a JSON string instead of a list, return empty."""
|
||||
self._set_series_rules_raw("[]")
|
||||
result = CoreSettings.get_dvr_series_rules()
|
||||
self.assertEqual(result, [])
|
||||
|
||||
def test_none_value_returns_empty(self):
|
||||
self._set_series_rules_raw(None)
|
||||
result = CoreSettings.get_dvr_series_rules()
|
||||
self.assertEqual(result, [])
|
||||
|
||||
def test_mixed_corrupt_elements(self):
|
||||
self._set_series_rules_raw([42, None, True, {"tvg_id": "x", "mode": "new", "title": "T"}])
|
||||
result = CoreSettings.get_dvr_series_rules()
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertEqual(result[0]["tvg_id"], "x")
|
||||
|
||||
|
||||
class SetDvrSeriesRulesTest(TestCase):
|
||||
"""Verify set_dvr_series_rules sanitizes input before persisting."""
|
||||
|
||||
def test_valid_rules_persisted(self):
|
||||
rules = [{"tvg_id": "abc", "mode": "all", "title": "Show"}]
|
||||
result = CoreSettings.set_dvr_series_rules(rules)
|
||||
self.assertEqual(result, rules)
|
||||
self.assertEqual(CoreSettings.get_dvr_series_rules(), rules)
|
||||
|
||||
def test_non_dict_elements_stripped_on_write(self):
|
||||
dirty = ["bad", 42, {"tvg_id": "abc", "mode": "all", "title": ""}]
|
||||
result = CoreSettings.set_dvr_series_rules(dirty)
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertEqual(result[0]["tvg_id"], "abc")
|
||||
self.assertEqual(CoreSettings.get_dvr_series_rules(), result)
|
||||
|
||||
def test_non_list_input_stores_empty(self):
|
||||
result = CoreSettings.set_dvr_series_rules("not a list")
|
||||
self.assertEqual(result, [])
|
||||
self.assertEqual(CoreSettings.get_dvr_series_rules(), [])
|
||||
|
||||
|
||||
class CoreSettingsSerializerDvrTest(TestCase):
|
||||
"""Verify the generic settings API sanitizes series_rules on save."""
|
||||
|
||||
def test_serializer_strips_corrupt_series_rules(self):
|
||||
"""Settings page round-trip must not persist corrupt series_rules."""
|
||||
from core.serializers import CoreSettingsSerializer
|
||||
|
||||
obj, _ = CoreSettings.objects.get_or_create(
|
||||
key=DVR_SETTINGS_KEY,
|
||||
defaults={"name": "DVR Settings", "value": {"series_rules": []}},
|
||||
)
|
||||
dirty_value = {
|
||||
**obj.value,
|
||||
"series_rules": ["bad", {"tvg_id": "ok", "mode": "all", "title": ""}],
|
||||
}
|
||||
serializer = CoreSettingsSerializer(obj, data={"value": dirty_value}, partial=True)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
serializer.save()
|
||||
obj.refresh_from_db()
|
||||
rules = obj.value.get("series_rules", [])
|
||||
self.assertEqual(len(rules), 1)
|
||||
self.assertEqual(rules[0]["tvg_id"], "ok")
|
||||
|
||||
def test_serializer_handles_non_list_series_rules(self):
|
||||
from core.serializers import CoreSettingsSerializer
|
||||
|
||||
obj, _ = CoreSettings.objects.get_or_create(
|
||||
key=DVR_SETTINGS_KEY,
|
||||
defaults={"name": "DVR Settings", "value": {"series_rules": []}},
|
||||
)
|
||||
dirty_value = {**obj.value, "series_rules": "not a list"}
|
||||
serializer = CoreSettingsSerializer(obj, data={"value": dirty_value}, partial=True)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
serializer.save()
|
||||
obj.refresh_from_db()
|
||||
self.assertEqual(obj.value.get("series_rules"), [])
|
||||
|
||||
|
||||
class EpgIgnoreListsTest(TestCase):
|
||||
"""Verify EPG ignore list getters handle corrupted stored data."""
|
||||
|
||||
def _set_epg_field_raw(self, field, raw_value):
|
||||
obj, _ = CoreSettings.objects.get_or_create(
|
||||
key=EPG_SETTINGS_KEY,
|
||||
defaults={"name": "EPG Settings", "value": {}},
|
||||
)
|
||||
current = obj.value if isinstance(obj.value, dict) else {}
|
||||
current[field] = raw_value
|
||||
obj.value = current
|
||||
obj.save()
|
||||
|
||||
def test_valid_string_lists_returned(self):
|
||||
for field, getter in [
|
||||
("epg_match_ignore_prefixes", CoreSettings.get_epg_match_ignore_prefixes),
|
||||
("epg_match_ignore_suffixes", CoreSettings.get_epg_match_ignore_suffixes),
|
||||
("epg_match_ignore_custom", CoreSettings.get_epg_match_ignore_custom),
|
||||
]:
|
||||
self._set_epg_field_raw(field, ["HD", "SD"])
|
||||
self.assertEqual(getter(), ["HD", "SD"])
|
||||
|
||||
def test_non_string_elements_filtered(self):
|
||||
for field, getter in [
|
||||
("epg_match_ignore_prefixes", CoreSettings.get_epg_match_ignore_prefixes),
|
||||
("epg_match_ignore_suffixes", CoreSettings.get_epg_match_ignore_suffixes),
|
||||
("epg_match_ignore_custom", CoreSettings.get_epg_match_ignore_custom),
|
||||
]:
|
||||
self._set_epg_field_raw(field, [42, None, "HD", True, "SD"])
|
||||
result = getter()
|
||||
self.assertEqual(result, ["HD", "SD"])
|
||||
|
||||
def test_non_list_value_returns_empty(self):
|
||||
for field, getter in [
|
||||
("epg_match_ignore_prefixes", CoreSettings.get_epg_match_ignore_prefixes),
|
||||
("epg_match_ignore_suffixes", CoreSettings.get_epg_match_ignore_suffixes),
|
||||
("epg_match_ignore_custom", CoreSettings.get_epg_match_ignore_custom),
|
||||
]:
|
||||
self._set_epg_field_raw(field, "not a list")
|
||||
self.assertEqual(getter(), [])
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue