diff --git a/core/models.py b/core/models.py index 4e6322e7..f1a604f7 100644 --- a/core/models.py +++ b/core/models.py @@ -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 diff --git a/core/serializers.py b/core/serializers.py index bff5f701..0df656b2 100644 --- a/core/serializers.py +++ b/core/serializers.py @@ -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 diff --git a/core/tests.py b/core/tests.py index 7ce503c2..305ab308 100644 --- a/core/tests.py +++ b/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(), [])