diff --git a/homeassistant/components/mqtt/mixins.py b/homeassistant/components/mqtt/mixins.py
index 9b1c7a9fb21..46c4a88e8fe 100644
--- a/homeassistant/components/mqtt/mixins.py
+++ b/homeassistant/components/mqtt/mixins.py
@@ -67,6 +67,25 @@ CONF_VIA_DEVICE = "via_device"
 CONF_DEPRECATED_VIA_HUB = "via_hub"
 CONF_SUGGESTED_AREA = "suggested_area"
 
+MQTT_ATTRIBUTES_BLOCKED = {
+    "assumed_state",
+    "available",
+    "context_recent_time",
+    "device_class",
+    "device_info",
+    "entity_picture",
+    "entity_registry_enabled_default",
+    "extra_state_attributes",
+    "force_update",
+    "icon",
+    "name",
+    "should_poll",
+    "state",
+    "supported_features",
+    "unique_id",
+    "unit_of_measurement",
+}
+
 MQTT_AVAILABILITY_SINGLE_SCHEMA = vol.Schema(
     {
         vol.Exclusive(CONF_AVAILABILITY_TOPIC, "availability"): valid_subscribe_topic,
@@ -175,11 +194,12 @@ async def async_setup_entry_helper(hass, domain, async_setup, schema):
 class MqttAttributes(Entity):
     """Mixin used for platforms that support JSON attributes."""
 
-    def __init__(self, config: dict) -> None:
+    def __init__(self, config: dict, extra_blocked_attributes: list = None) -> None:
         """Initialize the JSON attributes mixin."""
         self._attributes = None
         self._attributes_sub_state = None
         self._attributes_config = config
+        self._extra_blocked_attributes = extra_blocked_attributes or []
 
     async def async_added_to_hass(self) -> None:
         """Subscribe MQTT events."""
@@ -206,7 +226,13 @@ class MqttAttributes(Entity):
                     payload = attr_tpl.async_render_with_possible_json_value(payload)
                 json_dict = json.loads(payload)
                 if isinstance(json_dict, dict):
-                    self._attributes = json_dict
+                    filtered_dict = {
+                        k: v
+                        for k, v in json_dict.items()
+                        if k not in MQTT_ATTRIBUTES_BLOCKED
+                        and k not in self._extra_blocked_attributes
+                    }
+                    self._attributes = filtered_dict
                     self.async_write_ha_state()
                 else:
                     _LOGGER.warning("JSON result was not a dictionary")
diff --git a/tests/components/mqtt/test_common.py b/tests/components/mqtt/test_common.py
index fc8d26843d0..d8a263131a9 100644
--- a/tests/components/mqtt/test_common.py
+++ b/tests/components/mqtt/test_common.py
@@ -7,6 +7,7 @@ from unittest.mock import ANY, patch
 from homeassistant.components import mqtt
 from homeassistant.components.mqtt import debug_info
 from homeassistant.components.mqtt.const import MQTT_DISCONNECTED
+from homeassistant.components.mqtt.mixins import MQTT_ATTRIBUTES_BLOCKED
 from homeassistant.const import ATTR_ASSUMED_STATE, STATE_UNAVAILABLE
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.dispatcher import async_dispatcher_send
@@ -493,6 +494,37 @@ async def help_test_setting_attribute_via_mqtt_json_message(
     assert state.attributes.get("val") == "100"
 
 
+async def help_test_setting_blocked_attribute_via_mqtt_json_message(
+    hass, mqtt_mock, domain, config, extra_blocked_attributes
+):
+    """Test the setting of blocked attribute via MQTT with JSON payload.
+
+    This is a test helper for the MqttAttributes mixin.
+    """
+    extra_blocked_attributes = extra_blocked_attributes or []
+
+    # Add JSON attributes settings to config
+    config = copy.deepcopy(config)
+    config[domain]["json_attributes_topic"] = "attr-topic"
+    assert await async_setup_component(
+        hass,
+        domain,
+        config,
+    )
+    await hass.async_block_till_done()
+    val = "abc123"
+
+    for attr in MQTT_ATTRIBUTES_BLOCKED:
+        async_fire_mqtt_message(hass, "attr-topic", json.dumps({attr: val}))
+        state = hass.states.get(f"{domain}.test")
+        assert state.attributes.get(attr) != val
+
+    for attr in extra_blocked_attributes:
+        async_fire_mqtt_message(hass, "attr-topic", json.dumps({attr: val}))
+        state = hass.states.get(f"{domain}.test")
+        assert state.attributes.get(attr) != val
+
+
 async def help_test_setting_attribute_with_template(hass, mqtt_mock, domain, config):
     """Test the setting of attribute via MQTT with JSON payload.
 
diff --git a/tests/components/mqtt/test_sensor.py b/tests/components/mqtt/test_sensor.py
index 7d732849906..d90737a7b7f 100644
--- a/tests/components/mqtt/test_sensor.py
+++ b/tests/components/mqtt/test_sensor.py
@@ -42,6 +42,7 @@ from .test_common import (
     help_test_entity_id_update_subscriptions,
     help_test_setting_attribute_via_mqtt_json_message,
     help_test_setting_attribute_with_template,
+    help_test_setting_blocked_attribute_via_mqtt_json_message,
     help_test_unique_id,
     help_test_update_with_json_attrs_bad_JSON,
     help_test_update_with_json_attrs_not_dict,
@@ -531,6 +532,13 @@ async def test_setting_attribute_via_mqtt_json_message(hass, mqtt_mock):
     )
 
 
+async def test_setting_blocked_attribute_via_mqtt_json_message(hass, mqtt_mock):
+    """Test the setting of attribute via MQTT with JSON payload."""
+    await help_test_setting_blocked_attribute_via_mqtt_json_message(
+        hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG, None
+    )
+
+
 async def test_setting_attribute_with_template(hass, mqtt_mock):
     """Test the setting of attribute via MQTT with JSON payload."""
     await help_test_setting_attribute_with_template(